app/models/pipeline_invocation.rb

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