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