replay.rb

#!/usr/bin/env ruby
def check_require(name)
  begin
    gem name    
  rescue Gem::LoadError
    # not installed
    system("gem install #{name}")
    Gem.clear_paths
  end
  require name
end

check_require 'typhoeus'
check_require 'slop'
check_require 'json'
check_require 'ruby-metrics'
require 'ruby-metrics/integration/webrick'
check_require 'formatador'

opts = Slop.parse do
  banner 'Usage: replay.rb [options]'
  
  on :l, '--logs=', 'Collector logs path'
  on :c, '--concurrency=', 'Maximum concurrency', as: Integer
  on :t, '--tracker=', 'Collector tracker url'
  on :d, '--domain=', 'Domain', as: String, argument: :optional
end

class Replayer  
  def initialize(opts={})
    @formatador = Formatador.new
    @formatador.display_line('[green]Replay Collector Log[/]')    
    @concurrency = opts[:concurrency] ||= 10
    @tracker = opts[:tracker]         ||= "http://localhost:44080/tracker"
    @domain = "videocloud"
    @domain = opts[:domain] if opts[:domain]
    @path = File.absolute_path(opts[:logs])
    @hydra = Typhoeus::Hydra.new(max_concurrency: @concurrency)    
    @agent = Metrics::Agent.new
    @success_meter = @agent.meter :success_meter
    @success_counter = @agent.counter :success_counter
    @failed_meter  = @agent.meter :failed_meter
    @failed_counter = @agent.counter :failed_counter    
  end
  
  def process(file)
    @formatador.display_table([{:concurrency => @concurrency, :tracker_url => @tracker, :domain => @domain, :file => file.path}])
    count = 0
    @formatador.indent {
      file.each_slice(@concurrency) do |line_batch|
        line_batch.each do |line|
          count += 1
          params = JSON.parse(line).merge({"domain" => @domain})
          request = Typhoeus::Request.new(@tracker, method: :get, params: params)
          request.on_complete do |response|
            if response.success?
              @success_meter.mark
              @success_counter.incr
            else
              @failed_meter.mark
              @failed_counter.incr
            end
          end
          @hydra.queue request
          @formatador.redisplay("Request# %f \t [green]Success: %s[/] \t [red]Failure: %s[/] \t\t\t [green]Success Rate: %6.2f/min[/] \t [red]Failure Rate: %6.2f/min[/]" % [count, @success_counter, @failure_counter, @success_meter.one_minute_rate, @failed_meter.one_minute_rate])
        end
        # start sending out requests
        @hydra.run
        @formatador.redisplay("Request# %f \t [green]Success: %s[/] \t [red]Failure: %s[/] \t\t\t [green]Success Rate: %6.2f/min[/] \t [red]Failure Rate: %6.2f/min[/]" % [count, @success_counter, @failure_counter, @success_meter.one_minute_rate, @failed_meter.one_minute_rate])
      end      
    }
  end
  
  def get_file(path)
    file = nil
    if File.extname(@path) == ".gz"
      file = Zlib::GzipReader.new(open(@path)) 
    else
      file = open(@path)
    end
    return file
  end
  
  def run
    if File.file?(@path)
      if File.extname(@path) == ".gz"
        file = Zlib::GzipReader.new(open(@path)) 
      else
        file = open(@path)
      end      
      process(file) unless file.nil?
    else
      @formatador.display_line("[red]Path is not a file[/]")
    end
  end
end

Replayer.new(opts).run