<commit_before>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
<commit_msg>Ensure that Librato metric submission is threadsafe<commit_after>require 'librato/metrics'
require 'thread'

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)
      @mutex = Mutex.new
      add_listeners
    end

    private

    def add_listeners
      ActiveSupport::Notifications.subscribe('stream_reader.process_record') do |name, start, finish, id, payload|
        @mutex.synchronize do
          @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
end

{'fraction_non_alphanumeric': 0.12410785091197463, 'fraction_numerical': 0.0055511498810467885, 'mean_word_length': 2.4140730717185384, 'pattern_counts': {'":': 4, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4424537', 'n_tokens_mistral': 763, 'n_tokens_neox': 745, 'n_words': 119}