services/api/app/models/pipeline_instance.rb

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