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