dynflow message throughput monitor

dynflow_throughput_monitor.rb

# usage
#
#   1) copy the monitory to be loaded by satellite
#
#      cp dynflow_throughput_monitor.rb /usr/share/foreman/config/initializers
#      service foreman-tasks restart
#
#   2) grab the collected data to be shared with engineering:
#
#      tar cvf dynflow_thoughput_monitor.tar /var/log/foreman/dynflow_throughput_monitor.log*
#
#   3) remove the monitor after done collecting the data
#
#      rm /usr/share/foreman/config/initializers/dynflow_throughput_monitor.rb
#      service foreman-tasks restart
#
#   4) (engineering after receiving the collected tar) extract the logs:
#
#      gzip -d *.gz
#
#   5) extract relevant data:
#
#      * cat dynflow_thoughput_monitor.log* | sort -t';' -k2  > dynflow_throughput_monitor.csv
#

class DynflowThroughputMonitor < Dynflow::Middleware
  @logger = ::Logger.new(File.join(Rails.root, 'log', 'dynflow_throughput_monitor.log'))

  class << self
    attr_accessor :logger

    def log_header(pools)
      unless @log_stats_printed
        pools_columns = pools.map { |pool_name| "#{pool_name} queue_size; #{pool_name} free_workers"}.join(";")
        logger.info("dynflow_throughput_monitor;_sample_time;uuid;action_class;method;event;worker;time_start;time_end;duration;#{pools_columns}")
      end
      @log_stats_printed = true
    end

    def log_stats(sample_time, uuid, action_class, method, event, worker, time_start, time_end, duration, pools_data)

      pool_info = pools_data.keys.sort.map do |pool_name|
        pool_data = pools_data[pool_name]
        "#{pool_data[:jobs_size]};#{pool_data[:free_workers]}"
      end.join(';')
      logger.info("dynflow_throughput_monitor;#{sample_time};#{uuid};#{action_class};#{method};#{event};#{worker};#{time_start};#{time_end};#{duration};#{pool_info}")
    end
  end

  def actor_context(reference)
    reference.instance_variable_get('@core').instance_variable_get('@context')
  end

  def pools_data
    executor = ForemanTasks.dynflow.world.executor
    executor_core = actor_context(executor.instance_variable_get('@core'))
    pools = executor_core.instance_variable_get('@pools')
    unless pools
      pools = {:normal => executor_core.instance_variable_get('@pool')}
    end
    pools.inject({}) do |hash, (pool_name, pool)|
      pool = actor_context(pool)
      jobs = pool.instance_variable_get('@jobs').instance_variable_get('@jobs')
      free_workers = pool.instance_variable_get('@free_workers')
      hash.update(pool_name => { :jobs_size => jobs.size,
                                 :free_workers => free_workers.size })
    end
  end

  def run(*args)
    monitor_throughput('run', args.first) { pass(*args) }
  end

  def finalize
    monitor_throughput('finalize') { pass }
  end

  def monitor_throughput(method, event = nil)
    started_at = Time.now
    pools_data = self.pools_data
    uuid = "#{$$}/#{action.execution_plan_id}##{action.id}"
    yield
  ensure
    ended_at = Time.now
    self.class.log_header(pools_data.keys.sort)
    self.class.log_stats(format_time(started_at), uuid, action.class, method, event.inspect, Concurrent::Actor.current.name, format_time(started_at), format_time(ended_at), '%.10f' % (ended_at - started_at), pools_data)
  end

  def format_time(time)
    time.utc.strftime('%Y-%d-%m %H:%M:%S.%6N')
  end
end

ForemanTasks.dynflow.config.on_init do |world|
  world.middleware.use(DynflowThroughputMonitor)
end

Thanks for info: the problem was I was testing it against non-patched version, where the code was a bit different, should work now