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_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 :success
  22     t.add :active
  23     t.add :dependencies
  24     t.add :properties
  25     t.add :state
  26     t.add :components_summary
  27   end
  28
  29   # Supported states for a pipeline instance
  30   New = 'New'
  31   Ready = 'Ready'
  32   RunningOnServer = 'RunningOnServer'
  33   RunningOnClient = 'RunningOnClient'
  34   Paused = 'Paused'
  35   Failed = 'Failed'
  36   Complete = 'Complete'
  37
  38   def dependencies
  39     dependency_search(self.components).keys
  40   end
  41
  42   # if all components have input, the pipeline is Ready
  43   def components_look_ready?
  44     if !self.components || self.components.empty?
  45       return false
  46     end
  47
  48     all_components_have_input = true
  49     self.components.each do |name, component|
  50       component['script_parameters'].each do |parametername, parameter|
  51         parameter = { 'value' => parameter } unless parameter.is_a? Hash
  52         if parameter['value'].nil? and 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 self.components_look_ready?
 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, Paused
 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         self.state = Failed   # before_validation will fail if false is returned in the previous line
 175       when Complete
 176         self.active = false
 177         self.success = true
 178       else
 179         return false
 180       end
 181     elsif components_changed?
 182       if !self.state || self.state == New || !self.active
 183         if self.components_look_ready?
 184           self.state = Ready
 185         else
 186           self.state = New
 187         end
 188       end
 189     end
 190   end
 191
 192   def set_state_before_save
 193     if !self.state || self.state == New
 194       if self.active
 195         self.state = RunningOnServer
 196       elsif self.components_look_ready?
 197         self.state = Ready
 198       else
 199         self.state = New
 200       end
 201     end
 202   end
 203
 204 end