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