services/api/app/models/pipeline_instance.rb

   1 class PipelineInstance < ArvadosModel
   2   include AssignUuid
   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_success
  12   before_create :verify_status
  13   before_save :verify_status
  14
  15   api_accessible :user, extend: :common do |t|
  16     t.add :pipeline_template_uuid
  17     t.add :pipeline_template, :if => :pipeline_template
  18     t.add :name
  19     t.add :components
  20     t.add :success
  21     t.add :active
  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   New = 'New'
  30   Ready = 'Ready'
  31   RunningOnServer = 'RunningOnServer'
  32   RunningOnClient = 'RunningOnClient'
  33   Paused = 'Paused'
  34   Failed = 'Failed'
  35   Complete = 'Complete'
  36
  37   def dependencies
  38     dependency_search(self.components).keys
  39   end
  40
  41   # if all components have input, the pipeline is Ready
  42   def self.is_ready components
  43     if !components || components.empty?  # is this correct?
  44       return true
  45     end
  46
  47     all_components_have_input = true
  48     components.each do |name, component|
  49       component['script_parameters'].each do |parametername, parameter|
  50         parameter = { 'value' => parameter } unless parameter.is_a? Hash
  51         if parameter['value'].nil? and
  52             ![false,'false',0,'0'].index parameter['required']
  53           if parameter['output_of']
  54             next
  55           end
  56           all_components_have_input = false
  57           break
  58         end
  59       end
  60     end
  61     return all_components_have_input
  62   end
  63
  64   def progress_table
  65     begin
  66       # v0 pipeline format
  67       nrow = -1
  68       components['steps'].collect do |step|
  69         nrow += 1
  70         row = [nrow, step['name']]
  71         if step['complete'] and step['complete'] != 0
  72           if step['output_data_locator']
  73             row << 1.0
  74           else
  75             row << 0.0
  76           end
  77         else
  78           row << 0.0
  79           if step['failed']
  80             self.success = false
  81           end
  82         end
  83         row << (step['warehousejob']['id'] rescue nil)
  84         row << (step['warehousejob']['revision'] rescue nil)
  85         row << step['output_data_locator']
  86         row << (Time.parse(step['warehousejob']['finishtime']) rescue nil)
  87         row
  88       end
  89     rescue
  90       []
  91     end
  92   end
  93
  94   def progress_ratio
  95     t = progress_table
  96     return 0 if t.size < 1
  97     t.collect { |r| r[2] }.inject(0.0) { |sum,a| sum += a } / t.size
  98   end
  99
 100   def self.queue
 101     self.where('active = true')
 102   end
 103
 104   protected
 105   def bootstrap_components
 106     if pipeline_template and (!components or components.empty?)
 107       self.components = pipeline_template.components.deep_dup
 108     end
 109   end
 110
 111   def update_success
 112     if components and progress_ratio == 1.0
 113       self.success = true
 114     end
 115   end
 116
 117   def dependency_search(haystack)
 118     if haystack.is_a? String
 119       if (re = haystack.match /^([0-9a-f]{32}(\+[^,]+)*)+/)
 120         {re[1] => true}
 121       else
 122         {}
 123       end
 124     elsif haystack.is_a? Array
 125       deps = {}
 126       haystack.each do |value|
 127         deps.merge! dependency_search(value)
 128       end
 129       deps
 130     elsif haystack.respond_to? :keys
 131       deps = {}
 132       haystack.each do |key, value|
 133         deps.merge! dependency_search(value)
 134       end
 135       deps
 136     else
 137       {}
 138     end
 139   end
 140
 141   def verify_status
 142     if active_changed?
 143       if self.active
 144         self.state = RunningOnServer
 145       else
 146         if PipelineInstance.is_ready self.components
 147           self.state = Ready
 148         else
 149           self.state = New
 150         end
 151       end
 152     elsif success_changed?
 153       if self.success
 154         self.active = false
 155         self.state = Complete
 156       else
 157         self.active = false
 158         self.state = Failed
 159       end
 160     elsif state_changed?
 161       case self.state
 162       when New, Ready
 163         self.active = false
 164         self.success = nil
 165       when RunningOnServer
 166         self.active = true
 167         self.success = nil
 168       when RunningOnClient
 169         self.active = false
 170         self.success = nil
 171       when Failed
 172         self.active = false
 173         self.success = false
 174       when Complete
 175         self.active = false
 176         self.success = true
 177       end
 178     else    # new object create or save
 179       if !self.state || self.state == New || !self.active
 180         if PipelineInstance.is_ready self.components
 181           self.state = Ready
 182         else
 183           self.state = New
 184         end
 185       end
 186     end
 187   end
 188
 189 end