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   belongs_to :pipeline_template, :foreign_key => :pipeline_template_uuid, :primary_key => :uuid
   8   attr_accessor :pipeline_template
   9
  10   before_validation :bootstrap_components
  11   before_validation :update_success
  12
  13   api_accessible :user, extend: :common do |t|
  14     t.add :pipeline_template_uuid
  15     t.add :pipeline_template, :if => :pipeline_template
  16     t.add :name
  17     t.add :components
  18     t.add :success
  19     t.add :active
  20     t.add :dependencies
  21     t.add :properties
  22   end
  23
  24   def dependencies
  25     dependency_search(self.components).keys
  26   end
  27
  28   def progress_table
  29     begin
  30       # v0 pipeline format
  31       nrow = -1
  32       components['steps'].collect do |step|
  33         nrow += 1
  34         row = [nrow, step['name']]
  35         if step['complete'] and step['complete'] != 0
  36           if step['output_data_locator']
  37             row << 1.0
  38           else
  39             row << 0.0
  40           end
  41         else
  42           row << 0.0
  43           if step['failed']
  44             self.success = false
  45           end
  46         end
  47         row << (step['warehousejob']['id'] rescue nil)
  48         row << (step['warehousejob']['revision'] rescue nil)
  49         row << step['output_data_locator']
  50         row << (Time.parse(step['warehousejob']['finishtime']) rescue nil)
  51         row
  52       end
  53     rescue
  54       []
  55     end
  56   end
  57
  58   def progress_ratio
  59     t = progress_table
  60     return 0 if t.size < 1
  61     t.collect { |r| r[2] }.inject(0.0) { |sum,a| sum += a } / t.size
  62   end
  63
  64   def self.queue
  65     self.where('active = true')
  66   end
  67
  68   protected
  69   def bootstrap_components
  70     if pipeline_template and (!components or components.empty?)
  71       self.components = pipeline_template.components
  72     end
  73   end
  74
  75   def update_success
  76     if components and progress_ratio == 1.0
  77       self.success = true
  78     end
  79   end
  80
  81   def dependency_search(haystack)
  82     if haystack.is_a? String
  83       if (re = haystack.match /^([0-9a-f]{32}(\+[^,]+)*)+/)
  84         {re[1] => true}
  85       else
  86         {}
  87       end
  88     elsif haystack.is_a? Array
  89       deps = {}
  90       haystack.each do |value|
  91         deps.merge! dependency_search(value)
  92       end
  93       deps
  94     elsif haystack.respond_to? :keys
  95       deps = {}
  96       haystack.each do |key, value|
  97         deps.merge! dependency_search(value)
  98       end
  99       deps
 100     else
 101       {}
 102     end
 103   end
 104 end