#
# fluent-plugin-metricsense
#
# Copyright (C) 2012 Sadayuki Furuhashi
#
#    Licensed under the Apache License, Version 2.0 (the "License");
#    you may not use this file except in compliance with the License.
#    You may obtain a copy of the License at
#
#        http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS,
#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#    See the License for the specific language governing permissions and
#    limitations under the License.
#
module Fluent::MetricSenseOutput::Backends

  class DatadogBackend < Fluent::MetricSenseOutput::Backend
    Fluent::MetricSenseOutput.register_backend('datadog', self)

    config_param :dd_api_key, :string
    config_param :dd_app_key, :string, :default => nil
    config_param :host, :string, :default => nil
    config_param :device, :string, :default => nil
    config_param :timeout, :integer, :default => 5
    config_param :tags, :array, :default => []
    config_param :batch_size, :integer, :default => 200

    def initialize()
      super
      require "dogapi"
    end

    def configure(conf)
      super

      if @dd_api_key.nil?
        raise Fluent::ConfigError, "missing Datadog API key"
      end

      silent = false # raise exceptions on dogapi errors
      @dog = Dogapi::Client.new(@dd_api_key, @dd_app_key, @host, @device, silent, @timeout)
    end

    def write(data)
      data.each_slice(@batch_size) do |slice|
        metric_points = {}
        slice.each do |tag, time, value, seg_key, seg_val|
          if seg_key and seg_val
            # segmented values
            metric = "#{tag}_by_#{seg_key}"
            segment = "#{seg_key}:#{seg_val}"
          else
            # simple values
            metric = tag
            segment = ""
          end
          metric_points[metric] ||= {}
          metric_points[metric][segment] ||= []
          metric_points[metric][segment].push([Time.at(time), value])
        end

        metric_points.each do |metric, segment_points|
          segment_points.each do |segment, points|
            tags = @tags.dup
            if segment and not segment.empty?
              tags.push(segment)
            end

            options = {}
            options[:tags] = tags
            options[:host] = @host if @host
            options[:type] = "gauge"

            log.debug("datadog emit points: metric=#{metric}, points=#{points.inspect}, options=#{options.inspect}")
            begin
              code, response = @dog.emit_points(metric, points, options)
            rescue Exception => error
              # dogapi may raise an Exception.
              # fluentd expects StandardError as retriable error, though.
              raise("datadog error: #{error.class}: #{error.message}")
            end
            if code.to_i / 100 != 2
              raise("datadog error: HTTP #{code}: #{response.inspect}")
            end
          end
        end
      end
    end
  end
end