services/api/app/models/job.rb

   1 # Copyright (C) The Arvados Authors. All rights reserved.
   2 #
   3 # SPDX-License-Identifier: AGPL-3.0
   4 #
   5 #
   6 # Legacy jobs API aka crunch v1
   7 #
   8 # This is superceded by containers / container_requests (aka crunch v2)
   9 #
  10 # Arvados installations since the end of 2017 should have never
  11 # used jobs, and are unaffected by this change.
  12 #
  13 # So that older Arvados sites don't lose access to legacy records, the
  14 # API has been converted to read-only.  Creating and updating jobs
  15 # (and related types job_task, pipeline_template and
  16 # pipeline_instance) is disabled and much of the business logic
  17 # related has been removed, along with the crunch-dispatch.rb and
  18 # various other code specific to the jobs API.
  19 #
  20 # If you need to resurrect any of this code, here is the last commit
  21 # on master before the branch removing jobs API support:
  22 #
  23 # Wed Aug 7 14:49:38 2019 -0400 07d92519438a592d531f2c7558cd51788da262ca
  24
  25 require 'log_reuse_info'
  26 require 'safe_json'
  27
  28 class Job < ArvadosModel
  29   include HasUuid
  30   include KindAndEtag
  31   include CommonApiTemplate
  32   extend CurrentApiClient
  33   extend LogReuseInfo
  34   serialize :components, Hash
  35   serialize :script_parameters, Hash
  36   serialize :runtime_constraints, Hash
  37   serialize :tasks_summary, Hash
  38   before_create :ensure_unique_submit_id
  39   before_validation :set_priority
  40   before_validation :update_state_from_old_state_attrs
  41   before_validation :update_script_parameters_digest
  42   validate :ensure_script_version_is_commit
  43   validate :find_docker_image_locator
  44   validate :find_arvados_sdk_version
  45   validate :validate_status
  46   validate :validate_state_change
  47   validate :ensure_no_collection_uuids_in_script_params
  48   before_save :tag_version_in_internal_repository
  49   before_save :update_timestamps_when_state_changes
  50   before_create :create_disabled
  51   before_update :update_disabled
  52
  53   has_many(:nodes, foreign_key: 'job_uuid', primary_key: 'uuid')
  54
  55   class SubmitIdReused < RequestError
  56   end
  57
  58   api_accessible :user, extend: :common do |t|
  59     t.add :submit_id
  60     t.add :priority
  61     t.add :script
  62     t.add :script_parameters
  63     t.add :script_version
  64     t.add :cancelled_at
  65     t.add :cancelled_by_client_uuid
  66     t.add :cancelled_by_user_uuid
  67     t.add :started_at
  68     t.add :finished_at
  69     t.add :output
  70     t.add :success
  71     t.add :running
  72     t.add :state
  73     t.add :is_locked_by_uuid
  74     t.add :log
  75     t.add :runtime_constraints
  76     t.add :tasks_summary
  77     t.add :nondeterministic
  78     t.add :repository
  79     t.add :supplied_script_version
  80     t.add :arvados_sdk_version
  81     t.add :docker_image_locator
  82     t.add :queue_position
  83     t.add :node_uuids
  84     t.add :description
  85     t.add :components
  86   end
  87
  88   # Supported states for a job
  89   States = [
  90             (Queued = 'Queued'),
  91             (Running = 'Running'),
  92             (Cancelled = 'Cancelled'),
  93             (Failed = 'Failed'),
  94             (Complete = 'Complete'),
  95            ]
  96
  97   after_initialize do
  98     @need_crunch_dispatch_trigger = false
  99   end
 100
 101   def self.limit_index_columns_read
 102     ["components"]
 103   end
 104
 105   def self.protected_attributes
 106     [:arvados_sdk_version, :docker_image_locator]
 107   end
 108
 109   def assert_finished
 110     update(finished_at: finished_at || db_current_time,
 111                       success: success.nil? ? false : success,
 112                       running: false)
 113   end
 114
 115   def node_uuids
 116     nodes.map(&:uuid)
 117   end
 118
 119   def self.queue
 120     self.where('state = ?', Queued).order('priority desc, created_at')
 121   end
 122
 123   def queue_position
 124     # We used to report this accurately, but the implementation made queue
 125     # API requests O(n**2) for the size of the queue.  See #8800.
 126     # We've soft-disabled it because it's not clear we even want this
 127     # functionality: now that we have Node Manager with support for multiple
 128     # node sizes, "queue position" tells you very little about when a job will
 129     # run.
 130     state == Queued ? 0 : nil
 131   end
 132
 133   def self.running
 134     self.where('running = ?', true).
 135       order('priority desc, created_at')
 136   end
 137
 138   def lock locked_by_uuid
 139     with_lock do
 140       unless self.state == Queued and self.is_locked_by_uuid.nil?
 141         raise AlreadyLockedError
 142       end
 143       self.state = Running
 144       self.is_locked_by_uuid = locked_by_uuid
 145       self.save!
 146     end
 147   end
 148
 149   def update_script_parameters_digest
 150     self.script_parameters_digest = self.class.sorted_hash_digest(script_parameters)
 151   end
 152
 153   def self.searchable_columns operator
 154     super - ["script_parameters_digest"]
 155   end
 156
 157   def self.full_text_searchable_columns
 158     super - ["script_parameters_digest"]
 159   end
 160
 161   def self.load_job_specific_filters attrs, orig_filters, read_users
 162     # Convert Job-specific @filters entries into general SQL filters.
 163     script_info = {"repository" => nil, "script" => nil}
 164     git_filters = Hash.new do |hash, key|
 165       hash[key] = {"max_version" => "HEAD", "exclude_versions" => []}
 166     end
 167     filters = []
 168     orig_filters.each do |attr, operator, operand|
 169       if (script_info.has_key? attr) and (operator == "=")
 170         if script_info[attr].nil?
 171           script_info[attr] = operand
 172         elsif script_info[attr] != operand
 173           raise ArgumentError.new("incompatible #{attr} filters")
 174         end
 175       end
 176       case operator
 177       when "in git"
 178         git_filters[attr]["min_version"] = operand
 179       when "not in git"
 180         git_filters[attr]["exclude_versions"] += Array.wrap(operand)
 181       when "in docker", "not in docker"
 182         image_hashes = Array.wrap(operand).flat_map do |search_term|
 183           image_search, image_tag = search_term.split(':', 2)
 184           Collection.
 185             find_all_for_docker_image(image_search, image_tag, read_users, filter_compatible_format: false).
 186             map(&:portable_data_hash)
 187         end
 188         filters << [attr, operator.sub(/ docker$/, ""), image_hashes]
 189       else
 190         filters << [attr, operator, operand]
 191       end
 192     end
 193
 194     # Build a real script_version filter from any "not? in git" filters.
 195     git_filters.each_pair do |attr, filter|
 196       case attr
 197       when "script_version"
 198         script_info.each_pair do |key, value|
 199           if value.nil?
 200             raise ArgumentError.new("script_version filter needs #{key} filter")
 201           end
 202         end
 203         filter["repository"] = script_info["repository"]
 204         if attrs[:script_version]
 205           filter["max_version"] = attrs[:script_version]
 206         else
 207           # Using HEAD, set earlier by the hash default, is fine.
 208         end
 209       when "arvados_sdk_version"
 210         filter["repository"] = "arvados"
 211       else
 212         raise ArgumentError.new("unknown attribute for git filter: #{attr}")
 213       end
 214       revisions = CommitsHelper::find_commit_range(filter["repository"],
 215                                            filter["min_version"],
 216                                            filter["max_version"],
 217                                            filter["exclude_versions"])
 218       if revisions.empty?
 219         raise ArgumentError.
 220           new("error searching #{filter['repository']} from " +
 221               "'#{filter['min_version']}' to '#{filter['max_version']}', " +
 222               "excluding #{filter['exclude_versions']}")
 223       end
 224       filters.append([attr, "in", revisions])
 225     end
 226
 227     filters
 228   end
 229
 230   def self.default_git_filters(attr_name, repo_name, refspec)
 231     # Add a filter to @filters for `attr_name` = the latest commit available
 232     # in `repo_name` at `refspec`.  No filter is added if refspec can't be
 233     # resolved.
 234     commits = CommitsHelper::find_commit_range(repo_name, nil, refspec, nil)
 235     if commit_hash = commits.first
 236       [[attr_name, "=", commit_hash]]
 237     else
 238       []
 239     end
 240   end
 241
 242   def cancel(cascade: false, need_transaction: true)
 243     raise "No longer supported"
 244   end
 245
 246   protected
 247
 248   def self.sorted_hash_digest h
 249     Digest::MD5.hexdigest(Oj.dump(deep_sort_hash(h)))
 250   end
 251
 252   def foreign_key_attributes
 253     super + %w(output log)
 254   end
 255
 256   def skip_uuid_read_permission_check
 257     super + %w(cancelled_by_client_uuid)
 258   end
 259
 260   def skip_uuid_existence_check
 261     super + %w(output log)
 262   end
 263
 264   def set_priority
 265     if self.priority.nil?
 266       self.priority = 0
 267     end
 268     true
 269   end
 270
 271   def ensure_script_version_is_commit
 272     if state == Running
 273       # Apparently client has already decided to go for it. This is
 274       # needed to run a local job using a local working directory
 275       # instead of a commit-ish.
 276       return true
 277     end
 278     if new_record? or repository_changed? or script_version_changed?
 279       sha1 = CommitsHelper::find_commit_range(repository,
 280                                       nil, script_version, nil).first
 281       if not sha1
 282         errors.add :script_version, "#{script_version} does not resolve to a commit"
 283         return false
 284       end
 285       if supplied_script_version.nil? or supplied_script_version.empty?
 286         self.supplied_script_version = script_version
 287       end
 288       self.script_version = sha1
 289     end
 290     true
 291   end
 292
 293   def tag_version_in_internal_repository
 294     if state == Running
 295       # No point now. See ensure_script_version_is_commit.
 296       true
 297     elsif errors.any?
 298       # Won't be saved, and script_version might not even be valid.
 299       true
 300     elsif new_record? or repository_changed? or script_version_changed?
 301       uuid_was = uuid
 302       begin
 303         assign_uuid
 304         CommitsHelper::tag_in_internal_repository repository, script_version, uuid
 305       rescue
 306         self.uuid = uuid_was
 307         raise
 308       end
 309     end
 310   end
 311
 312   def ensure_unique_submit_id
 313     if !submit_id.nil?
 314       if Job.where('submit_id=?',self.submit_id).first
 315         raise SubmitIdReused.new
 316       end
 317     end
 318     true
 319   end
 320
 321   def resolve_runtime_constraint(key, attr_sym)
 322     if ((runtime_constraints.is_a? Hash) and
 323         (search = runtime_constraints[key]))
 324       ok, result = yield search
 325     else
 326       ok, result = true, nil
 327     end
 328     if ok
 329       send("#{attr_sym}=".to_sym, result)
 330     else
 331       errors.add(attr_sym, result)
 332     end
 333     ok
 334   end
 335
 336   def find_arvados_sdk_version
 337     resolve_runtime_constraint("arvados_sdk_version",
 338                                :arvados_sdk_version) do |git_search|
 339       commits = CommitsHelper::find_commit_range("arvados",
 340                                          nil, git_search, nil)
 341       if commits.empty?
 342         [false, "#{git_search} does not resolve to a commit"]
 343       elsif not runtime_constraints["docker_image"]
 344         [false, "cannot be specified without a Docker image constraint"]
 345       else
 346         [true, commits.first]
 347       end
 348     end
 349   end
 350
 351   def find_docker_image_locator
 352     if runtime_constraints.is_a? Hash and Rails.configuration.Containers.JobsAPI.DefaultDockerImage != ""
 353       runtime_constraints['docker_image'] ||=
 354         Rails.configuration.Containers.JobsAPI.DefaultDockerImage
 355     end
 356
 357     resolve_runtime_constraint("docker_image",
 358                                :docker_image_locator) do |image_search|
 359       image_tag = runtime_constraints['docker_image_tag']
 360       if coll = Collection.for_latest_docker_image(image_search, image_tag)
 361         [true, coll.portable_data_hash]
 362       else
 363         [false, "not found for #{image_search}"]
 364       end
 365     end
 366   end
 367
 368   def permission_to_update
 369     if is_locked_by_uuid_was and !(current_user and
 370                                    (current_user.uuid == is_locked_by_uuid_was or
 371                                     current_user.uuid == system_user.uuid))
 372       if script_changed? or
 373           script_parameters_changed? or
 374           script_version_changed? or
 375           (!cancelled_at_was.nil? and
 376            (cancelled_by_client_uuid_changed? or
 377             cancelled_by_user_uuid_changed? or
 378             cancelled_at_changed?)) or
 379           started_at_changed? or
 380           finished_at_changed? or
 381           running_changed? or
 382           success_changed? or
 383           output_changed? or
 384           log_changed? or
 385           tasks_summary_changed? or
 386           (state_changed? && state != Cancelled) or
 387           components_changed?
 388         logger.warn "User #{current_user.uuid if current_user} tried to change protected job attributes on locked #{self.class.to_s} #{uuid_was}"
 389         return false
 390       end
 391     end
 392     if !is_locked_by_uuid_changed?
 393       super
 394     else
 395       if !current_user
 396         logger.warn "Anonymous user tried to change lock on #{self.class.to_s} #{uuid_was}"
 397         false
 398       elsif is_locked_by_uuid_was and is_locked_by_uuid_was != current_user.uuid
 399         logger.warn "User #{current_user.uuid} tried to steal lock on #{self.class.to_s} #{uuid_was} from #{is_locked_by_uuid_was}"
 400         false
 401       elsif !is_locked_by_uuid.nil? and is_locked_by_uuid != current_user.uuid
 402         logger.warn "User #{current_user.uuid} tried to lock #{self.class.to_s} #{uuid_was} with uuid #{is_locked_by_uuid}"
 403         false
 404       else
 405         super
 406       end
 407     end
 408   end
 409
 410   def update_modified_by_fields
 411     if self.cancelled_at_changed?
 412       # Ensure cancelled_at cannot be set to arbitrary non-now times,
 413       # or changed once it is set.
 414       if self.cancelled_at and not self.cancelled_at_was
 415         self.cancelled_at = db_current_time
 416         self.cancelled_by_user_uuid = current_user.uuid
 417         self.cancelled_by_client_uuid = current_api_client.andand.uuid
 418         @need_crunch_dispatch_trigger = true
 419       else
 420         self.cancelled_at = self.cancelled_at_was
 421         self.cancelled_by_user_uuid = self.cancelled_by_user_uuid_was
 422         self.cancelled_by_client_uuid = self.cancelled_by_client_uuid_was
 423       end
 424     end
 425     super
 426   end
 427
 428   def update_timestamps_when_state_changes
 429     return if not (state_changed? or new_record?)
 430
 431     case state
 432     when Running
 433       self.started_at ||= db_current_time
 434     when Failed, Complete
 435       self.finished_at ||= db_current_time
 436     when Cancelled
 437       self.cancelled_at ||= db_current_time
 438     end
 439
 440     # TODO: Remove the following case block when old "success" and
 441     # "running" attrs go away. Until then, this ensures we still
 442     # expose correct success/running flags to older clients, even if
 443     # some new clients are writing only the new state attribute.
 444     case state
 445     when Queued
 446       self.running = false
 447       self.success = nil
 448     when Running
 449       self.running = true
 450       self.success = nil
 451     when Cancelled, Failed
 452       self.running = false
 453       self.success = false
 454     when Complete
 455       self.running = false
 456       self.success = true
 457     end
 458     self.running ||= false # Default to false instead of nil.
 459
 460     @need_crunch_dispatch_trigger = true
 461
 462     true
 463   end
 464
 465   def update_state_from_old_state_attrs
 466     # If a client has touched the legacy state attrs, update the
 467     # "state" attr to agree with the updated values of the legacy
 468     # attrs.
 469     #
 470     # TODO: Remove this method when old "success" and "running" attrs
 471     # go away.
 472     if cancelled_at_changed? or
 473         success_changed? or
 474         running_changed? or
 475         state.nil?
 476       if cancelled_at
 477         self.state = Cancelled
 478       elsif success == false
 479         self.state = Failed
 480       elsif success == true
 481         self.state = Complete
 482       elsif running == true
 483         self.state = Running
 484       else
 485         self.state = Queued
 486       end
 487     end
 488     true
 489   end
 490
 491   def validate_status
 492     if self.state.in?(States)
 493       true
 494     else
 495       errors.add :state, "#{state.inspect} must be one of: #{States.inspect}"
 496       false
 497     end
 498   end
 499
 500   def validate_state_change
 501     ok = true
 502     if self.state_changed?
 503       ok = case self.state_was
 504            when nil
 505              # state isn't set yet
 506              true
 507            when Queued
 508              # Permit going from queued to any state
 509              true
 510            when Running
 511              # From running, may only transition to a finished state
 512              [Complete, Failed, Cancelled].include? self.state
 513            when Complete, Failed, Cancelled
 514              # Once in a finished state, don't permit any more state changes
 515              false
 516            else
 517              # Any other state transition is also invalid
 518              false
 519            end
 520       if not ok
 521         errors.add :state, "invalid change from #{self.state_was} to #{self.state}"
 522       end
 523     end
 524     ok
 525   end
 526
 527   def ensure_no_collection_uuids_in_script_params
 528     # Fail validation if any script_parameters field includes a string containing a
 529     # collection uuid pattern.
 530     if self.script_parameters_changed?
 531       if recursive_hash_search(self.script_parameters, Collection.uuid_regex)
 532         self.errors.add :script_parameters, "must use portable_data_hash instead of collection uuid"
 533         return false
 534       end
 535     end
 536     true
 537   end
 538
 539   # recursive_hash_search searches recursively through hashes and
 540   # arrays in 'thing' for string fields matching regular expression
 541   # 'pattern'.  Returns true if pattern is found, false otherwise.
 542   def recursive_hash_search thing, pattern
 543     if thing.is_a? Hash
 544       thing.each do |k, v|
 545         return true if recursive_hash_search v, pattern
 546       end
 547     elsif thing.is_a? Array
 548       thing.each do |k|
 549         return true if recursive_hash_search k, pattern
 550       end
 551     elsif thing.is_a? String
 552       return true if thing.match pattern
 553     end
 554     false
 555   end
 556
 557   def create_disabled
 558     raise "Disabled"
 559   end
 560
 561   def update_disabled
 562     raise "Disabled"
 563   end
 564 end