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