Feature/facter 2/fact 65 aggregate resolutions

lib/facter/core/aggregate.rb

@@ -0,0 +1,191 @@
+require 'facter'
+require 'facter/core/directed_graph'
+require 'facter/core/suitable'
+require 'facter/core/resolvable'
+
+# Aggregates provide a mechanism for facts to be resolved in multiple steps.
+#
+# Aggregates are evaluated in two parts: generating individual chunks and then
+# aggregating all chunks together. Each chunk is a block of code that generates
+# a value, and may depend on other chunks when it runs. After all chunks have
+# been evaluated they are passed to the aggregate block as Hash<name, result>.
+# The aggregate block converts the individual chunks into a single value that is
+# returned as the final value of the aggregate.
+#
+# @api public
+# @since 2.0.0
+class Facter::Core::Aggregate
+
+  include Facter::Core::Suitable
+  include Facter::Core::Resolvable
+
+  # @!attribute [r] name
+  #   @return [Symbol] The name of the aggregate resolution
+  attr_reader :name
+
+  # @!attribute [r] deps
+  #   @api private
+  #   @return [Facter::Core::DirectedGraph]
+  attr_reader :deps
+
+  # @!attribute [r] confines
+  #   @return [Array<Facter::Core::Confine>] An array of confines restricting
+  #     this to a specific platform
+  #   @see Facter::Core::Suitable
+  attr_reader :confines
+
+  def initialize(name)
+    @name = name
+
+    @confines = []
+    @chunks = {}
+
+    @aggregate = nil
+    @deps = Facter::Core::DirectedGraph.new
+  end
+
+  def set_options(options)
+    if options[:name]
+      @name = options.delete(:name)
+    end
+
+    if options.has_key?(:timeout)
+      @timeout = options.delete(:timeout)
+    end
+
+    if options.has_key?(:weight)
+      @weight = options.delete(:weight)
+    end
+
+    if not options.keys.empty?
+      raise ArgumentError, "Invalid aggregate options #{options.keys.inspect}"
+    end
+  end
+
+  # Define a new chunk for the given aggregate
+  #
+  # @api public
+  #
+  # @example Defining a chunk with no dependencies
+  #   aggregate.chunk(:mountpoints) do
+  #     # generate mountpoint information
+  #   end
+  #
+  # @example Defining an chunk to add mount options
+  #   aggregate.chunk(:mount_options, :require => [:mountpoints]) do |mountpoints|
+  #     # `mountpoints` is the result of the previous chunk
+  #     # generate mount option information based on the mountpoints
+  #   end
+  #
+  # @param name [Symbol] A name unique to this aggregate describing the chunk
+  # @param opts [Hash]
+  # @options opts [Array<Symbol>, Symbol] :require One or more chunks
+  #   to evaluate and pass to this block.
+  # @yield [*Object] Zero or more chunk results
+  #
+  # @return [void]
+  def chunk(name, opts = {}, &block)
+    if not block_given?
+      raise ArgumentError, "#{self.class.name}#chunk requires a block"
+    end
+
+    deps = Array(opts.delete(:require))
+
+    if not opts.empty?
+      raise ArgumentError, "Unexpected options passed to #{self.class.name}#chunk: #{opts.keys.inspect}"
+    end
+
+    @deps[name] = deps
+    @chunks[name] = block
+  end
+
+  # Define how all chunks should be combined
+  #
+  # @api public
+  #
+  # @example Merge all chunks
+  #   aggregate.aggregate do |chunks|
+  #     final_result = {}
+  #     chunks.each_value do |chunk|
+  #       final_result.deep_merge(chunk)
+  #     end
+  #     final_result
+  #   end
+  #
+  # @example Sum all chunks
+  #   aggregate.aggregate do |chunks|
+  #     total = 0
+  #     chunks.each_value do |chunk|
+  #       total += chunk
+  #     end
+  #     total
+  #   end
+  #
+  # @yield [Hash<Symbol, Object>] A hash containing chunk names and
+  #   chunk values
+  #
+  # @return [void]
+  def aggregate(&block)
+    if block_given?
+      @aggregate = block
+    else
+      raise ArgumentError, "#{self.class.name}#aggregate requires a block"
+    end
+  end
+
+  private
+
+  # Evaluate the results of this aggregate.
+  #
+  # @see Facter::Core::Resolvable#value
+  # @return [Object]
+  def resolve_value
+    chunk_results = run_chunks()
+    aggregate_results(chunk_results)
+  end
+
+  # Order all chunks based on their dependencies and evaluate each one, passing
+  # dependent chunks as needed.
+  #
+  # @return [Hash<Symbol, Object>] A hash containing the chunk that
+  #   generated value and the related value.
+  def run_chunks
+    results = {}
+    order_chunks.each do |(name, block)|
+      begin
+        input = @deps[name].map { |dep_name| results[dep_name] }
+
+        results[name] = block.call(*input)
+      rescue => e
+        Facter.warn "Could not run chunk #{name}:#{block}: #{e.message}"
+      end
+    end
+
+    results
+  end
+
+  # Process the results of all chunks with the aggregate block and return the results.
+  # @return [Object]
+  def aggregate_results(results)
+    @aggregate.call(results)
+  rescue => e
+    Facter.warn "Could not aggregate chunks for #{name}: #{e.message}"
+  end
+
+  # Order chunks based on their dependencies
+  #
+  # @return [Array<Symbol, Proc>] A list of chunk names and blocks in evaluation order.
+  def order_chunks
+    if not @deps.acyclic?
+      raise DependencyError, "Could not order chunks; found the following dependency cycles: #{@deps.cycles.inspect}"
+    end
+
+    sorted_names = @deps.tsort
+
+    sorted_names.map do |name|
+      [name, @chunks[name]]
+    end
+  end
+
+  class DependencyError < StandardError; end
+end

lib/facter/core/directed_graph.rb

@@ -0,0 +1,45 @@
+require 'set'
+
+module Facter
+  module Core
+    class DirectedGraph < Hash
+      include TSort
+
+      def acyclic?
+        cycles.empty?
+      end
+
+      def cycles
+        cycles = []
+        each_strongly_connected_component do |component|
+          cycles << component if component.size > 1
+        end
+        cycles
+      end
+
+      alias tsort_each_node each_key
+
+      def tsort_each_child(node)
+        fetch(node, []).each do |child|
+          yield child
+        end
+      end
+
+      def tsort
+        missing = Set.new(self.values.flatten) - Set.new(self.keys)
+
+        if not missing.empty?
+          raise MissingVertex, "Cannot sort elements; cannot depend on missing elements #{missing.to_a}"
+        end
+
+        super
+
+      rescue TSort::Cyclic
+        raise CycleError, "Cannot sort elements; found the following cycles: #{cycles.inspect}"
+      end
+
+      class CycleError < StandardError; end
+      class MissingVertex < StandardError; end
+    end
+  end
+end

lib/facter/core/resolvable.rb

@@ -0,0 +1,95 @@
+require 'timeout'
+
+# The resolvable mixin defines behavior for evaluating and returning fact
+# resolutions.
+#
+# Classes including this mixin should implement at #name method describing
+# the value being resolved and a #resolve_value that actually executes the code
+# to resolve the value.
+module Facter::Core::Resolvable
+
+  # The timeout, in seconds, for evaluating this resolution.
+  # @return [Integer]
+  # @api public
+  attr_accessor :timeout
+
+  # Return the timeout period for resolving a value.
+  # (see #timeout)
+  # @return [Numeric]
+  # @comment requiring 'timeout' stdlib class causes Object#timeout to be
+  #   defined which delegates to Timeout.timeout. This method may potentially
+  #   overwrite the #timeout attr_reader on this class, so we define #limit to
+  #   avoid conflicts.
+  def limit
+    @timeout || 0
+  end
+
+  ##
+  # on_flush accepts a block and executes the block when the resolution's value
+  # is flushed.  This makes it possible to model a single, expensive system
+  # call inside of a Ruby object and then define multiple dynamic facts which
+  # resolve by sending messages to the model instance.  If one of the dynamic
+  # facts is flushed then it can, in turn, flush the data stored in the model
+  # instance to keep all of the dynamic facts in sync without making multiple,
+  # expensive, system calls.
+  #
+  # Please see the Solaris zones fact for an example of how this feature may be
+  # used.
+  #
+  # @see Facter::Util::Fact#flush
+  # @see Facter::Util::Resolution#flush
+  #
+  # @api public
+  def on_flush(&block)
+    @on_flush_block = block
+  end
+
+  ##
+  # flush executes the block, if any, stored by the {on_flush} method
+  #
+  # @see Facter::Util::Fact#flush
+  # @see Facter::Util::Resolution#on_flush
+  #
+  # @api private
+  def flush
+    @on_flush_block.call if @on_flush_block
+  end
+
+  def value
+    result = nil
+
+    with_timing do
+      Timeout.timeout(limit) do
+        result = resolve_value
+      end
+    end
+
+    Facter::Util::Normalization.normalize(result)
+  rescue Timeout::Error => detail
+    Facter.warn "Timed out seeking value for #{self.name}"
+
+    # This call avoids zombies -- basically, create a thread that will
+    # dezombify all of the child processes that we're ignoring because
+    # of the timeout.
+    Thread.new { Process.waitall }
+    return nil
+  rescue Facter::Util::Normalization::NormalizationError => e
+    Facter.warn "Fact resolution #{self.name} resolved to an invalid value: #{e.message}"
+    return nil
+  rescue => details
+    Facter.warn "Could not retrieve #{self.name}: #{details.message}"
+    return nil
+  end
+
+  private
+
+  def with_timing
+    starttime = Time.now.to_f
+
+    yield
+
+    finishtime = Time.now.to_f
+    ms = (finishtime - starttime) * 1000
+    Facter.show_time "#{self.name}: #{"%.2f" % ms}ms"
+  end
+end