services/api/app/models/job.rb

   1 class Job < ArvadosModel
   2   include HasUuid
   3   include KindAndEtag
   4   include CommonApiTemplate
   5   serialize :script_parameters, Hash
   6   serialize :runtime_constraints, Hash
   7   serialize :tasks_summary, Hash
   8   before_create :ensure_unique_submit_id
   9   before_create :ensure_script_version_is_commit
  10   before_update :ensure_script_version_is_commit
  11   after_commit :trigger_crunch_dispatch_if_cancelled, :on => :update
  12
  13   has_many :commit_ancestors, :foreign_key => :descendant, :primary_key => :script_version
  14
  15   class SubmitIdReused < StandardError
  16   end
  17
  18   api_accessible :user, extend: :common do |t|
  19     t.add :submit_id
  20     t.add :priority
  21     t.add :script
  22     t.add :script_parameters
  23     t.add :script_version
  24     t.add :cancelled_at
  25     t.add :cancelled_by_client_uuid
  26     t.add :cancelled_by_user_uuid
  27     t.add :started_at
  28     t.add :finished_at
  29     t.add :output
  30     t.add :output_is_persistent
  31     t.add :success
  32     t.add :running
  33     t.add :is_locked_by_uuid
  34     t.add :log
  35     t.add :runtime_constraints
  36     t.add :tasks_summary
  37     t.add :dependencies
  38     t.add :log_stream_href
  39     t.add :log_buffer
  40     t.add :nondeterministic
  41     t.add :repository
  42     t.add :supplied_script_version
  43   end
  44
  45   def assert_finished
  46     update_attributes(finished_at: finished_at || Time.now,
  47                       success: success.nil? ? false : success,
  48                       running: false)
  49   end
  50
  51   def log_stream_href
  52     unless self.finished_at
  53       "#{current_api_base}/#{self.class.to_s.pluralize.underscore}/#{self.uuid}/log_tail_follow"
  54     end
  55   end
  56
  57   def self.queue
  58     self.where('started_at is ? and is_locked_by_uuid is ? and cancelled_at is ? and success is ?',
  59                nil, nil, nil, nil).
  60       order('priority desc, created_at')
  61   end
  62
  63   def self.running
  64     self.where('running = ?', true).
  65       order('priority desc, created_at')
  66   end
  67
  68   protected
  69
  70   def foreign_key_attributes
  71     super + %w(output log)
  72   end
  73
  74   def skip_uuid_read_permission_check
  75     super + %w(cancelled_by_client_uuid)
  76   end
  77
  78   def skip_uuid_existence_check
  79     super + %w(output log)
  80   end
  81
  82   def ensure_script_version_is_commit
  83     if self.is_locked_by_uuid and self.started_at
  84       # Apparently client has already decided to go for it. This is
  85       # needed to run a local job using a local working directory
  86       # instead of a commit-ish.
  87       return true
  88     end
  89     if new_record? or script_version_changed?
  90       sha1 = Commit.find_commit_range(current_user, self.repository, nil, self.script_version, nil)[0] rescue nil
  91       if sha1
  92         self.supplied_script_version = self.script_version if self.supplied_script_version.nil? or self.supplied_script_version.empty?
  93         self.script_version = sha1
  94       else
  95         raise ArgumentError.new("Specified script_version does not resolve to a commit")
  96       end
  97     end
  98   end
  99
 100   def ensure_unique_submit_id
 101     if !submit_id.nil?
 102       if Job.where('submit_id=?',self.submit_id).first
 103         raise SubmitIdReused.new
 104       end
 105     end
 106     true
 107   end
 108
 109   def dependencies
 110     deps = {}
 111     queue = self.script_parameters.values
 112     while not queue.empty?
 113       queue = queue.flatten.compact.collect do |v|
 114         if v.is_a? Hash
 115           v.values
 116         elsif v.is_a? String
 117           v.match(/^(([0-9a-f]{32})\b(\+[^,]+)?,?)*$/) do |locator|
 118             deps[locator.to_s] = true
 119           end
 120           nil
 121         end
 122       end
 123     end
 124     deps.keys
 125   end
 126
 127   def permission_to_update
 128     if is_locked_by_uuid_was and !(current_user and
 129                                    (current_user.uuid == is_locked_by_uuid_was or
 130                                     current_user.uuid == system_user.uuid))
 131       if script_changed? or
 132           script_parameters_changed? or
 133           script_version_changed? or
 134           (!cancelled_at_was.nil? and
 135            (cancelled_by_client_uuid_changed? or
 136             cancelled_by_user_uuid_changed? or
 137             cancelled_at_changed?)) or
 138           started_at_changed? or
 139           finished_at_changed? or
 140           running_changed? or
 141           success_changed? or
 142           output_changed? or
 143           log_changed? or
 144           tasks_summary_changed?
 145         logger.warn "User #{current_user.uuid if current_user} tried to change protected job attributes on locked #{self.class.to_s} #{uuid_was}"
 146         return false
 147       end
 148     end
 149     if !is_locked_by_uuid_changed?
 150       super
 151     else
 152       if !current_user
 153         logger.warn "Anonymous user tried to change lock on #{self.class.to_s} #{uuid_was}"
 154         false
 155       elsif is_locked_by_uuid_was and is_locked_by_uuid_was != current_user.uuid
 156         logger.warn "User #{current_user.uuid} tried to steal lock on #{self.class.to_s} #{uuid_was} from #{is_locked_by_uuid_was}"
 157         false
 158       elsif !is_locked_by_uuid.nil? and is_locked_by_uuid != current_user.uuid
 159         logger.warn "User #{current_user.uuid} tried to lock #{self.class.to_s} #{uuid_was} with uuid #{is_locked_by_uuid}"
 160         false
 161       else
 162         super
 163       end
 164     end
 165   end
 166
 167   def update_modified_by_fields
 168     if self.cancelled_at_changed?
 169       # Ensure cancelled_at cannot be set to arbitrary non-now times,
 170       # or changed once it is set.
 171       if self.cancelled_at and not self.cancelled_at_was
 172         self.cancelled_at = Time.now
 173         self.cancelled_by_user_uuid = current_user.uuid
 174         self.cancelled_by_client_uuid = current_api_client.uuid
 175         @need_crunch_dispatch_trigger = true
 176       else
 177         self.cancelled_at = self.cancelled_at_was
 178         self.cancelled_by_user_uuid = self.cancelled_by_user_uuid_was
 179         self.cancelled_by_client_uuid = self.cancelled_by_client_uuid_was
 180       end
 181     end
 182     super
 183   end
 184
 185   def trigger_crunch_dispatch_if_cancelled
 186     if @need_crunch_dispatch_trigger
 187       File.open(Rails.configuration.crunch_refresh_trigger, 'wb') do
 188         # That's all, just create/touch a file for crunch-job to see.
 189       end
 190     end
 191   end
 192
 193   def log_buffer
 194     begin
 195       @@redis ||= Redis.new(:timeout => 0)
 196       if @@redis.exists uuid
 197         @@redis.getrange(uuid, 0 - 2**10, -1)
 198       end
 199     rescue Redis::CannotConnectError
 200       return '(not available)'
 201     end
 202   end
 203 end