require 'librato/metrics'

module LibratoReporter
  class << self
    def run!
      @client = Librato::Metrics::Client.new
      @client.authenticate(ENV['LIBRATO_USER'],
                           ENV['LIBRATO_TOKEN'])
      autosubmit_interval = Integer(ENV['LIBRATO_AUTOSUBMIT_INTERVAL'] || 30)
      @queue = Librato::Metrics::Queue.new(autosubmit_interval: autosubmit_interval,
                                           client: @client)
      add_listeners
    end

    private

    def add_listeners
      ActiveSupport::Notifications.subscribe('stream_reader.process_record') do |name, start, finish, id, payload|
        @queue.add "#{name}.duration": {
          value: (finish - start),
          source: "#{payload[:stream_name]}:#{payload[:prefix]}:#{payload[:shard_id]}" },
          "#{name}.latency": {
          value: payload[:ms_behind],
          source: "#{payload[:stream_name]}:#{payload[:prefix]}:#{payload[:shard_id]}" }
      end
    end
  end
end

require 'librato/metrics'

module LibratoReporter
  class << self
    def run!
      @client = Librato::Metrics::Client.new
      @client.authenticate(ENV['LIBRATO_USER'],
                           ENV['LIBRATO_TOKEN'])
      autosubmit_interval = Integer(ENV['LIBRATO_AUTOSUBMIT_INTERVAL'] || 60)
      autosubmit_count    = Integer(ENV['LIBRATO_AUTOSUBMIT_COUNT'] || 10000)
      @aggregator = Librato::Metrics::Aggregator.new(autosubmit_interval: autosubmit_interval,
                                                     autosubmit_count: autosubmit_count,
                                                     client: @client)
      add_listeners
    end

    private

    def add_listeners
      ActiveSupport::Notifications.subscribe('stream_reader.process_record') do |name, start, finish, id, payload|
        @aggregator.add "#{name}.duration": {
          value: (finish - start),
          source: "#{payload[:stream_name]}:#{payload[:prefix]}:#{payload[:shard_id]}" },
          "#{name}.latency": {
          value: payload[:ms_behind],
          source: "#{payload[:stream_name]}:#{payload[:prefix]}:#{payload[:shard_id]}" }
      end
    end
  end
end

Switch to using a Librato aggregator instead of a queue
Switch to using a Librato aggregator instead of a queue

From their docs (https://github.com/librato/librato-metrics#aggregate-measurements)
seems like this may be more appropriate for the kind of usage we have
here.

Also enable automatic submission based on count and tune it via
`LIBRATO_AUTOSUBMIT_COUNT`.
