<commit_before>require 'eventmachine'
require 'em-http'

require_relative '../logging'
require_relative '../plugin'
require_relative '../reporter'

module EVD::Plugin::KairosDB
  include EVD::Plugin
  include EVD::Logging

  register_plugin "kairosdb"

  class Output
    include EVD::Reporter

    set_reporter_keys :dropped_metrics, :sent_metrics, :failed_metrics

    HEADER = {
      "Content-Type" => "application/json"
    }

    API_PATH = "/api/v1/datapoints"

    def initialize log, opts={}
      @log = log
      @url = opts[:url]
      @flush_interval = opts[:flush_interval]
      @buffer_limit = opts[:buffer_limit]
      @buffer = []
      @sub = nil
      @pending = nil
      @conn = nil
    end

    def id
      "kairosdb_http_output-#{@url}"
    end

    def flush!
      if @timer
        @timer.cancel
        @timer = nil
      end

      if @pending
        @log.info "Request already in progress, dropping metrics"
        increment :dropped_metrics, @buffer.size
        @buffer = []
        return
      end

      unless @conn
        @log.error "No active connection"
        increment :dropped_metrics, @buffer.size
        @buffer = []
        return
      end

      events = make_events(@buffer)
      buffer_size = @buffer.size
      @buffer = []

      events = JSON.dump(events)

      @log.info "Sending metrics to #{@url}"

      @pending = @conn.post(
        :path => API_PATH, :head => HEADER, :body => events)

      @pending.callback do
        increment :sent_metrics, buffer_size
        @pending = nil
      end

      @pending.errback do
        @log.error "Failed to submit events: #{@pending.error.exception}"
        increment :failed_metrics, buffer_size
        @pending = nil
      end
    end

    # optimized, makes the assumption that all events have the same metadata as
    # the first seen one.
    def make_events buffer
      groups = {}

      buffer.each do |metric|
        group = (groups[metric.key] ||= {
          :name => make_name(metric.key), :tags => make_tags(metric),
          :datapoints => []})
        group[:datapoints] << [(metric.time.to_f * 1000).to_i, metric.value]
      end

      return groups.values
    end

    def make_name key
      key = key.gsub " ", "/"
      key.gsub ":", "_"
    end

    def make_tags metric
      tags = {
        "host" => metric.host
      }

      metric.attributes.each do |key, value|
        tags[key] = value
      end

      metric.tags.each do |tag|
        tags[tag] = ""
      end

      return tags
    end

    def check_timer!
      return if @timer

      @log.debug "Setting timer to #{@flush_interval}s"

      @timer = EM::Timer.new(@flush_interval) do
        flush!
      end
    end

    def start output
      @log.info "Will send events to #{@url}"

      @conn = EM::HttpRequest.new(@url)

      @sub = output.metric_subscribe do |metric|<commit_msg>kairosdb: Do not include tags as kairosdb-tags with empty values, the datapoints are silently rejected.<commit_after>require 'eventmachine'
require 'em-http'

require_relative '../logging'
require_relative '../plugin'
require_relative '../reporter'

module EVD::Plugin::KairosDB
  include EVD::Plugin
  include EVD::Logging

  register_plugin "kairosdb"

  class Output
    include EVD::Reporter

    set_reporter_keys :dropped_metrics, :sent_metrics, :failed_metrics

    HEADER = {
      "Content-Type" => "application/json"
    }

    API_PATH = "/api/v1/datapoints"

    def initialize log, opts={}
      @log = log
      @url = opts[:url]
      @flush_interval = opts[:flush_interval]
      @buffer_limit = opts[:buffer_limit]
      @buffer = []
      @sub = nil
      @pending = nil
      @conn = nil
    end

    def id
      "kairosdb_http_output-#{@url}"
    end

    def flush!
      if @timer
        @timer.cancel
        @timer = nil
      end

      if @pending
        @log.info "Request already in progress, dropping metrics"
        increment :dropped_metrics, @buffer.size
        @buffer = []
        return
      end

      unless @conn
        @log.error "No active connection"
        increment :dropped_metrics, @buffer.size
        @buffer = []
        return
      end

      events = make_events(@buffer)
      buffer_size = @buffer.size
      @buffer = []

      events = JSON.dump(events)

      @log.info "Sending metrics to #{@url}"

      @pending = @conn.post(
        :path => API_PATH, :head => HEADER, :body => events)

      @pending.callback do
        increment :sent_metrics, buffer_size
        @pending = nil
      end

      @pending.errback do
        @log.error "Failed to submit events: #{@pending.error.exception}"
        increment :failed_metrics, buffer_size
        @pending = nil
      end
    end

    # optimized, makes the assumption that all events have the same metadata as
    # the first seen one.
    def make_events buffer
      groups = {}

      buffer.each do |metric|
        group = (groups[metric.key] ||= {
          :name => make_name(metric.key), :tags => make_tags(metric),
          :datapoints => []})
        group[:datapoints] << [(metric.time.to_f * 1000).to_i, metric.value]
      end

      return groups.values
    end

    # Warning: These are the 'bad' characters I've been able to reverse
    # engineer so far.
    def make_name key
      key = key.gsub " ", "/"
      key.gsub ":", "_"
    end

    # Warning: KairosDB ignores complete metrics if you use tags which have no
    # values, therefore I have not figured out a way to transport 'tags'.
    def make_tags metric
      tags = {
        "host" => metric.host
      }

      metric.attributes.each do |key, value|
        tags[key] = value
      end

      return tags
    end

    def check_timer!
      return if @timer

      @log.debug "Setting timer to #{@flush_interval}s"

      @timer = EM::Timer.new(@flush_interval) do
        flush!
      end
    end

    def start output
      @log.info "Will send events to #{@url}"

      @conn = EM::HttpRequest.new(@url)

      @sub = output.metric_subscribe do |metric|
{'fraction_non_alphanumeric': 0.10925619834710744, 'fraction_numerical': 0.001652892561983471, 'mean_word_length': 2.1401141670991177, 'pattern_counts': {'":': 2, '<': 7, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6313639', 'n_tokens_mistral': 1885, 'n_tokens_neox': 1815, 'n_words': 506}