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