services/api/app/models/job.rb

   1 class Job < ArvadosModel
   2   include HasUuid
   3   include KindAndEtag
   4   include CommonApiTemplate
   5   attr_protected :docker_image_locator
   6   serialize :script_parameters, Hash
   7   serialize :runtime_constraints, Hash
   8   serialize :tasks_summary, Hash
   9   before_create :ensure_unique_submit_id
  10   after_commit :trigger_crunch_dispatch_if_cancelled, :on => :update
  11   before_validation :set_priority
  12   before_validation :update_timestamps_when_state_changes
  13   before_validation :update_state_from_old_state_attrs
  14   validate :ensure_script_version_is_commit
  15   validate :find_docker_image_locator
  16   validate :validate_status
  17
  18   has_many :commit_ancestors, :foreign_key => :descendant, :primary_key => :script_version
  19   has_many(:nodes, foreign_key: :job_uuid, primary_key: :uuid)
  20
  21   class SubmitIdReused < StandardError
  22   end
  23
  24   api_accessible :user, extend: :common do |t|
  25     t.add :submit_id
  26     t.add :priority
  27     t.add :script
  28     t.add :script_parameters
  29     t.add :script_version
  30     t.add :cancelled_at
  31     t.add :cancelled_by_client_uuid
  32     t.add :cancelled_by_user_uuid
  33     t.add :started_at
  34     t.add :finished_at
  35     t.add :output
  36     t.add :success
  37     t.add :running
  38     t.add :state
  39     t.add :is_locked_by_uuid
  40     t.add :log
  41     t.add :runtime_constraints
  42     t.add :tasks_summary
  43     t.add :dependencies
  44     t.add :nondeterministic
  45     t.add :repository
  46     t.add :supplied_script_version
  47     t.add :docker_image_locator
  48     t.add :queue_position
  49     t.add :node_uuids
  50     t.add :description
  51   end
  52
  53   # Supported states for a job
  54   States = [
  55             (Queued = 'Queued'),
  56             (Running = 'Running'),
  57             (Cancelled = 'Cancelled'),
  58             (Failed = 'Failed'),
  59             (Complete = 'Complete'),
  60            ]
  61
  62   def assert_finished
  63     update_attributes(finished_at: finished_at || Time.now,
  64                       success: success.nil? ? false : success,
  65                       running: false)
  66   end
  67
  68   def node_uuids
  69     nodes.map(&:uuid)
  70   end
  71
  72   def self.queue
  73     self.where('state = ?', Queued).order('priority desc, created_at')
  74   end
  75
  76   def queue_position
  77     i = 0
  78     Job::queue.each do |j|
  79       if j[:uuid] == self.uuid
  80         return i
  81       end
  82     end
  83     nil
  84   end
  85
  86   def self.running
  87     self.where('running = ?', true).
  88       order('priority desc, created_at')
  89   end
  90
  91   protected
  92
  93   def foreign_key_attributes
  94     super + %w(output log)
  95   end
  96
  97   def skip_uuid_read_permission_check
  98     super + %w(cancelled_by_client_uuid)
  99   end
 100
 101   def skip_uuid_existence_check
 102     super + %w(output log)
 103   end
 104
 105   def set_priority
 106     if self.priority.nil?
 107       self.priority = 0
 108     end
 109     true
 110   end
 111
 112   def ensure_script_version_is_commit
 113     if self.is_locked_by_uuid and self.started_at
 114       # Apparently client has already decided to go for it. This is
 115       # needed to run a local job using a local working directory
 116       # instead of a commit-ish.
 117       return true
 118     end
 119     if new_record? or script_version_changed?
 120       sha1 = Commit.find_commit_range(current_user, self.repository, nil, self.script_version, nil)[0] rescue nil
 121       if sha1
 122         self.supplied_script_version = self.script_version if self.supplied_script_version.nil? or self.supplied_script_version.empty?
 123         self.script_version = sha1
 124       else
 125         self.errors.add :script_version, "#{self.script_version} does not resolve to a commit"
 126         return false
 127       end
 128     end
 129   end
 130
 131   def ensure_unique_submit_id
 132     if !submit_id.nil?
 133       if Job.where('submit_id=?',self.submit_id).first
 134         raise SubmitIdReused.new
 135       end
 136     end
 137     true
 138   end
 139
 140   def find_docker_image_locator
 141     # Find the Collection that holds the Docker image specified in the
 142     # runtime constraints, and store its locator in docker_image_locator.
 143     unless runtime_constraints.is_a? Hash
 144       # We're still in validation stage, so we can't assume
 145       # runtime_constraints isn't something horrible like an array or
 146       # a string. Treat those cases as "no docker image supplied";
 147       # other validations will fail anyway.
 148       self.docker_image_locator = nil
 149       return true
 150     end
 151     image_search = runtime_constraints['docker_image']
 152     image_tag = runtime_constraints['docker_image_tag']
 153     if image_search.nil?
 154       self.docker_image_locator = nil
 155       true
 156     elsif coll = Collection.for_latest_docker_image(image_search, image_tag)
 157       self.docker_image_locator = coll.portable_data_hash
 158       true
 159     else
 160       errors.add(:docker_image_locator, "not found for #{image_search}")
 161       false
 162     end
 163   end
 164
 165   def dependencies
 166     deps = {}
 167     queue = self.script_parameters.values
 168     while not queue.empty?
 169       queue = queue.flatten.compact.collect do |v|
 170         if v.is_a? Hash
 171           v.values
 172         elsif v.is_a? String
 173           v.match(/^(([0-9a-f]{32})\b(\+[^,]+)?,?)*$/) do |locator|
 174             deps[locator.to_s] = true
 175           end
 176           nil
 177         end
 178       end
 179     end
 180     deps.keys
 181   end
 182
 183   def permission_to_update
 184     if is_locked_by_uuid_was and !(current_user and
 185                                    (current_user.uuid == is_locked_by_uuid_was or
 186                                     current_user.uuid == system_user.uuid))
 187       if script_changed? or
 188           script_parameters_changed? or
 189           script_version_changed? or
 190           (!cancelled_at_was.nil? and
 191            (cancelled_by_client_uuid_changed? or
 192             cancelled_by_user_uuid_changed? or
 193             cancelled_at_changed?)) or
 194           started_at_changed? or
 195           finished_at_changed? or
 196           running_changed? or
 197           success_changed? or
 198           output_changed? or
 199           log_changed? or
 200           tasks_summary_changed?
 201         logger.warn "User #{current_user.uuid if current_user} tried to change protected job attributes on locked #{self.class.to_s} #{uuid_was}"
 202         return false
 203       end
 204     end
 205     if !is_locked_by_uuid_changed?
 206       super
 207     else
 208       if !current_user
 209         logger.warn "Anonymous user tried to change lock on #{self.class.to_s} #{uuid_was}"
 210         false
 211       elsif is_locked_by_uuid_was and is_locked_by_uuid_was != current_user.uuid
 212         logger.warn "User #{current_user.uuid} tried to steal lock on #{self.class.to_s} #{uuid_was} from #{is_locked_by_uuid_was}"
 213         false
 214       elsif !is_locked_by_uuid.nil? and is_locked_by_uuid != current_user.uuid
 215         logger.warn "User #{current_user.uuid} tried to lock #{self.class.to_s} #{uuid_was} with uuid #{is_locked_by_uuid}"
 216         false
 217       else
 218         super
 219       end
 220     end
 221   end
 222
 223   def update_modified_by_fields
 224     if self.cancelled_at_changed?
 225       # Ensure cancelled_at cannot be set to arbitrary non-now times,
 226       # or changed once it is set.
 227       if self.cancelled_at and not self.cancelled_at_was
 228         self.cancelled_at = Time.now
 229         self.cancelled_by_user_uuid = current_user.uuid
 230         self.cancelled_by_client_uuid = current_api_client.andand.uuid
 231         @need_crunch_dispatch_trigger = true
 232       else
 233         self.cancelled_at = self.cancelled_at_was
 234         self.cancelled_by_user_uuid = self.cancelled_by_user_uuid_was
 235         self.cancelled_by_client_uuid = self.cancelled_by_client_uuid_was
 236       end
 237     end
 238     super
 239   end
 240
 241   def trigger_crunch_dispatch_if_cancelled
 242     if @need_crunch_dispatch_trigger
 243       File.open(Rails.configuration.crunch_refresh_trigger, 'wb') do
 244         # That's all, just create/touch a file for crunch-job to see.
 245       end
 246     end
 247   end
 248
 249   def update_timestamps_when_state_changes
 250     return if not (state_changed? or new_record?)
 251     case state
 252     when Running
 253       self.started_at ||= Time.now
 254     when Failed, Complete
 255       self.finished_at ||= Time.now
 256     when Cancelled
 257       self.cancelled_at ||= Time.now
 258     end
 259
 260     # TODO: Remove the following case block when old "success" and
 261     # "running" attrs go away. Until then, this ensures we still
 262     # expose correct success/running flags to older clients, even if
 263     # some new clients are writing only the new state attribute.
 264     case state
 265     when Queued
 266       self.running = false
 267       self.success = nil
 268     when Running
 269       self.running = true
 270       self.success = nil
 271     when Cancelled, Failed
 272       self.running = false
 273       self.success = false
 274     when Complete
 275       self.running = false
 276       self.success = true
 277     end
 278     self.running ||= false # Default to false instead of nil.
 279
 280     true
 281   end
 282
 283   def update_state_from_old_state_attrs
 284     # If a client has touched the legacy state attrs, update the
 285     # "state" attr to agree with the updated values of the legacy
 286     # attrs.
 287     #
 288     # TODO: Remove this method when old "success" and "running" attrs
 289     # go away.
 290     if cancelled_at_changed? or
 291         success_changed? or
 292         running_changed? or
 293         state.nil?
 294       if cancelled_at
 295         self.state = Cancelled
 296       elsif success == false
 297         self.state = Failed
 298       elsif success == true
 299         self.state = Complete
 300       elsif running == true
 301         self.state = Running
 302       else
 303         self.state = Queued
 304       end
 305     end
 306     true
 307   end
 308
 309   def validate_status
 310     if self.state.in?(States)
 311       true
 312     else
 313       errors.add :state, "#{state.inspect} must be one of: #{States.inspect}"
 314       false
 315     end
 316   end
 317
 318 end