iNecas/dynflow_throughput_monitor.rb

## 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
	# 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