services/api/app/models/pipeline_instance.rb

   1 # Copyright (C) The Arvados Authors. All rights reserved.
   2 #
   3 # SPDX-License-Identifier: AGPL-3.0
   4
   5 class PipelineInstance < ArvadosModel
   6   include HasUuid
   7   include KindAndEtag
   8   include CommonApiTemplate
   9   serialize :components, Hash
  10   serialize :properties, Hash
  11   serialize :components_summary, Hash
  12   belongs_to :pipeline_template,
  13              foreign_key: 'pipeline_template_uuid',
  14              primary_key: 'uuid',
  15              optional: true
  16
  17   before_validation :bootstrap_components
  18   before_validation :update_state
  19   before_validation :verify_status
  20   before_validation :update_timestamps_when_state_changes
  21   before_create :set_state_before_save
  22   before_save :set_state_before_save
  23   before_create :create_disabled
  24   before_update :update_disabled
  25
  26   api_accessible :user, extend: :common do |t|
  27     t.add :pipeline_template_uuid
  28     t.add :name
  29     t.add :components
  30     t.add :properties
  31     t.add :state
  32     t.add :components_summary
  33     t.add :description
  34     t.add :started_at
  35     t.add :finished_at
  36   end
  37
  38   # Supported states for a pipeline instance
  39   States =
  40     [
  41      (New = 'New'),
  42      (Ready = 'Ready'),
  43      (RunningOnServer = 'RunningOnServer'),
  44      (RunningOnClient = 'RunningOnClient'),
  45      (Paused = 'Paused'),
  46      (Failed = 'Failed'),
  47      (Complete = 'Complete'),
  48     ]
  49
  50   def self.limit_index_columns_read
  51     ["components"]
  52   end
  53
  54   # if all components have input, the pipeline is Ready
  55   def components_look_ready?
  56     if !self.components || self.components.empty?
  57       return false
  58     end
  59
  60     all_components_have_input = true
  61     self.components.each do |name, component|
  62       component['script_parameters'].andand.each do |parametername, parameter|
  63         parameter = { 'value' => parameter } unless parameter.is_a? Hash
  64         if parameter['value'].nil? and parameter['required']
  65           if parameter['output_of']
  66             next
  67           end
  68           all_components_have_input = false
  69           break
  70         end
  71       end
  72     end
  73     return all_components_have_input
  74   end
  75
  76   def progress_table
  77     begin
  78       # v0 pipeline format
  79       nrow = -1
  80       components['steps'].collect do |step|
  81         nrow += 1
  82         row = [nrow, step['name']]
  83         if step['complete'] and step['complete'] != 0
  84           if step['output_data_locator']
  85             row << 1.0
  86           else
  87             row << 0.0
  88           end
  89         else
  90           row << 0.0
  91           if step['failed']
  92             self.state = Failed
  93           end
  94         end
  95         row << (step['warehousejob']['id'] rescue nil)
  96         row << (step['warehousejob']['revision'] rescue nil)
  97         row << step['output_data_locator']
  98         row << (Time.parse(step['warehousejob']['finishtime']) rescue nil)
  99         row
 100       end
 101     rescue
 102       []
 103     end
 104   end
 105
 106   def progress_ratio
 107     t = progress_table
 108     return 0 if t.size < 1
 109     t.collect { |r| r[2] }.inject(0.0) { |sum,a| sum += a } / t.size
 110   end
 111
 112   def self.queue
 113     self.where("state = 'RunningOnServer'")
 114   end
 115
 116   def cancel(cascade: false, need_transaction: true)
 117     raise "No longer supported"
 118   end
 119
 120   protected
 121   def bootstrap_components
 122     if pipeline_template and (!components or components.empty?)
 123       self.components = pipeline_template.components.deep_dup
 124     end
 125   end
 126
 127   def update_state
 128     if components and progress_ratio == 1.0
 129       self.state = Complete
 130     end
 131   end
 132
 133   def verify_status
 134     changed_attributes = self.changed
 135
 136     if new_record? or 'components'.in? changed_attributes
 137       self.state ||= New
 138       if (self.state == New) and self.components_look_ready?
 139         self.state = Ready
 140       end
 141     end
 142
 143     if !self.state.in?(States)
 144       errors.add :state, "'#{state.inspect} must be one of: [#{States.join ', '}]"
 145       throw(:abort)
 146     end
 147   end
 148
 149   def set_state_before_save
 150     if self.components_look_ready? && (!self.state || self.state == New)
 151       self.state = Ready
 152     end
 153   end
 154
 155   def update_timestamps_when_state_changes
 156     return if not (state_changed? or new_record?)
 157
 158     case state
 159     when RunningOnServer, RunningOnClient
 160       self.started_at ||= db_current_time
 161     when Failed, Complete
 162       current_time = db_current_time
 163       self.started_at ||= current_time
 164       self.finished_at ||= current_time
 165     end
 166   end
 167
 168
 169   def create_disabled
 170     raise "Disabled"
 171   end
 172
 173   def update_disabled
 174     raise "Disabled"
 175   end
 176 end