first commit

.gitignore

@@ -0,0 +1,17 @@
+*.gem
+*.rbc
+.bundle
+.config
+.yardoc
+Gemfile.lock
+InstalledFiles
+_yardoc
+coverage
+doc/
+lib/bundler/man
+pkg
+rdoc
+spec/reports
+test/tmp
+test/version_tmp
+tmp

.rspec

@@ -0,0 +1,2 @@
+--require spec_helper
+--colour

Gemfile

@@ -0,0 +1,4 @@
+source 'https://rubygems.org'
+
+# Specify your gem's dependencies in fluent-plugin-dd.gemspec
+gemspec

LICENSE.txt

@@ -0,0 +1,22 @@
+Copyright (c) 2014 Genki Sugawara
+
+MIT License
+
+Permission is hereby granted, free of charge, to any person obtaining
+a copy of this software and associated documentation files (the
+"Software"), to deal in the Software without restriction, including
+without limitation the rights to use, copy, modify, merge, publish,
+distribute, sublicense, and/or sell copies of the Software, and to
+permit persons to whom the Software is furnished to do so, subject to
+the following conditions:
+
+The above copyright notice and this permission notice shall be
+included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
+LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

README.md

@@ -0,0 +1,31 @@
+# fluent-plugin-dd
+
+Output plugin for Datadog
+
+## Installation
+
+    $ gem install fluent-plugin-dd
+
+## Configuration
+
+```
+<match datadog.**>
+  type dd
+  dd_api_key ...
+  #host my_host.example.com
+</match>
+```
+
+## Usage
+
+```sh
+echo '{"metric":"some.metric.name", "value":50.0}' | fluent-cat datadog.metric
+```
+
+## Contributing
+
+1. Fork it
+2. Create your feature branch (`git checkout -b my-new-feature`)
+3. Commit your changes (`git commit -am 'Add some feature'`)
+4. Push to the branch (`git push origin my-new-feature`)
+5. Create new Pull Request

Rakefile

@@ -0,0 +1,6 @@
+require 'bundler/gem_tasks'
+require 'rspec/core/rake_task'
+
+RSpec::Core::RakeTask.new('spec')
+
+task :default => :spec

fluent-plugin-dd.gemspec

@@ -0,0 +1,22 @@
+# coding: utf-8
+Gem::Specification.new do |spec|
+  spec.name          = 'fluent-plugin-dd'
+  spec.version       = '0.1.0'
+  spec.authors       = ['Genki Sugawara']
+  spec.email         = ['sugawara@cookpad.com']
+  spec.description   = %q{Output plugin for Datadog}
+  spec.summary       = %q{Output plugin for Datadog}
+  spec.homepage      = 'https://bitbucket.org/winebarrel/fluent-plugin-dd'
+  spec.license       = 'MIT'
+
+  spec.files         = `git ls-files`.split($/)
+  spec.executables   = spec.files.grep(%r{^bin/}) { |f| File.basename(f) }
+  spec.test_files    = spec.files.grep(%r{^(test|spec|features)/})
+  spec.require_paths = ['lib']
+
+  spec.add_dependency 'fluentd'
+  spec.add_dependency 'dogapi'
+  spec.add_development_dependency 'bundler', '~> 1.3'
+  spec.add_development_dependency 'rake'
+  spec.add_development_dependency 'rspec', '>= 2.11.0'
+end

lib/fluent/plugin/out_dd.rb

@@ -0,0 +1,79 @@
+class Fluent::DdOutput < Fluent::BufferedOutput
+  include Fluent::SetTimeKeyMixin
+  include Fluent::SetTagKeyMixin
+
+  Fluent::Plugin.register_output('dd', self)
+
+  unless method_defined?(:log)
+    define_method('log') { $log }
+  end
+
+  config_set_default :include_time_key, true
+  config_set_default :include_tag_key, true
+
+  config_param :dd_api_key, :string
+  config_param :host, :string, :default => nil
+
+  def initialize
+    super
+    require 'dogapi'
+    require 'socket'
+    require 'time'
+  end
+
+  def start
+    super
+  end
+
+  def shutdown
+    super
+  end
+
+  def configure(conf)
+    super
+
+    unless @dd_api_key
+      raise Fluent::ConfigError, '`dd_api_key` is required'
+    end
+
+    unless @host
+      @host = %x[hostname -f 2> /dev/null].strip
+      @host = Socket.gethostname if @host.empty?
+    end
+
+    @dog = Dogapi::Client.new(@dd_api_key, nil, @host)
+  end
+
+  def format(tag, time, record)
+    record.to_msgpack
+  end
+
+  def write(chunk)
+    enum = chunk.to_enum(:msgpack_each)
+
+    enum.select {|record|
+      unless record['metric']
+        log.warn("`metric` key does not exist: #{record.inspect}")
+      end
+
+      record['metric']
+    }.chunk {|record|
+      record.values_at('metric', 'tag', 'host', 'type')
+    }.each {|i, records|
+      metric, tag, host, type = i
+
+      points = records.map do |record|
+        time = Time.parse(record['time'])
+        value = record['value']
+        [time, value]
+      end
+
+      options = {}
+      options['tags'] = [tag] if tag
+      options['host'] = host if host
+      options['type'] = type if type
+
+      @dog.emit_points(metric, points, options)
+    }
+  end
+end

spec/out_dd_spec.rb

@@ -0,0 +1,138 @@
+describe Fluent::DdOutput do
+  let(:time) {
+    Time.parse('2014-02-08 04:14:15 UTC').to_i
+  }
+
+  it 'should receive an API key' do
+    Dogapi::Client.should_receive(:new).with("test_dd_api_key", nil, "test_host")
+    run_driver {|d, dog| }
+  end
+
+  it 'should be called emit_points' do
+    run_driver do |d, dog|
+      dog.should_receive(:emit_points).with(
+        "some.metric.name",
+        [[Time.parse("2014-02-08 04:14:15 UTC"), 50.0],
+         [Time.parse("2014-02-08 04:14:15 UTC"), 100.0]],
+        {"tags"=>["test.default"]}
+      )
+
+      d.emit({"metric" => "some.metric.name", "value" => 50.0}, time)
+      d.emit({"metric" => "some.metric.name", "value" => 100.0}, time)
+    end
+  end
+
+  it 'should be called emit_points for each tag' do
+    run_driver do |d, dog|
+      dog.should_receive(:emit_points).with(
+        "some.metric.name",
+        [[Time.parse("2014-02-08 04:14:15 UTC"), 50.0],
+         [Time.parse("2014-02-08 04:14:15 UTC"), 100.0]],
+        {"tags"=>["test.1"]}
+      )
+
+      dog.should_receive(:emit_points).with(
+        "some.metric.name",
+        [[Time.parse("2014-02-08 04:14:15 UTC"), 150.0],
+         [Time.parse("2014-02-08 04:14:15 UTC"), 200.0]],
+        {"tags"=>["test.2"]}
+      )
+
+      dog.should_receive(:emit_points).with(
+        "some.metric.name",
+        [[Time.parse("2014-02-08 04:14:15 UTC"), 250.0],
+         [Time.parse("2014-02-08 04:14:15 UTC"), 300.0]],
+        {"tags"=>["test.3"]}
+      )
+
+      d.tag = 'test.1'
+      d.emit({"metric" => "some.metric.name", "value" => 50.0}, time)
+      d.emit({"metric" => "some.metric.name", "value" => 100.0}, time)
+
+      d.tag = 'test.2'
+      d.emit({"metric" => "some.metric.name", "value" => 150.0}, time)
+      d.emit({"metric" => "some.metric.name", "value" => 200.0}, time)
+
+      d.tag = 'test.3'
+      d.emit({"metric" => "some.metric.name", "value" => 250.0}, time)
+      d.emit({"metric" => "some.metric.name", "value" => 300.0}, time)
+    end
+  end
+
+  it 'should be called emit_points for each host' do
+    run_driver do |d, dog|
+      dog.should_receive(:emit_points).with(
+        "some.metric.name",
+        [[Time.parse("2014-02-08 04:14:15 UTC"), 50.0],
+         [Time.parse("2014-02-08 04:14:15 UTC"), 100.0]],
+        {"tags"=>["test.default"], "host"=>"www1.example.com"}
+      )
+
+      dog.should_receive(:emit_points).with(
+        "some.metric.name",
+        [[Time.parse("2014-02-08 04:14:15 UTC"), 150.0],
+         [Time.parse("2014-02-08 04:14:15 UTC"), 200.0]],
+        {"tags"=>["test.default"], "host"=>"www2.example.com"}
+      )
+
+      dog.should_receive(:emit_points).with(
+        "some.metric.name",
+        [[Time.parse("2014-02-08 04:14:15 UTC"), 250.0],
+         [Time.parse("2014-02-08 04:14:15 UTC"), 300.0]],
+        {"tags"=>["test.default"], "host"=>"www3.example.com"}
+      )
+
+      d.emit({"metric" => "some.metric.name", "value" => 50.0, "host" => "www1.example.com"}, time)
+      d.emit({"metric" => "some.metric.name", "value" => 100.0, "host" => "www1.example.com"}, time)
+      d.emit({"metric" => "some.metric.name", "value" => 150.0, "host" => "www2.example.com"}, time)
+      d.emit({"metric" => "some.metric.name", "value" => 200.0, "host" => "www2.example.com"}, time)
+      d.emit({"metric" => "some.metric.name", "value" => 250.0, "host" => "www3.example.com"}, time)
+      d.emit({"metric" => "some.metric.name", "value" => 300.0, "host" => "www3.example.com"}, time)
+    end
+  end
+
+  it 'should be called emit_points for each type' do
+    run_driver do |d, dog|
+      dog.should_receive(:emit_points).with(
+        "some.metric.name",
+        [[Time.parse("2014-02-08 04:14:15 UTC"), 50.0],
+         [Time.parse("2014-02-08 04:14:15 UTC"), 100.0]],
+        {"tags"=>["test.default"], "type"=>"gauge"}
+      )
+
+      dog.should_receive(:emit_points).with(
+        "some.metric.name",
+        [[Time.parse("2014-02-08 04:14:15 UTC"), 150.0],
+         [Time.parse("2014-02-08 04:14:15 UTC"), 200.0]],
+        {"tags"=>["test.default"], "type"=>"counter"}
+      )
+
+      d.emit({"metric" => "some.metric.name", "value" => 50.0, "type" => "gauge"}, time)
+      d.emit({"metric" => "some.metric.name", "value" => 100.0, "type" => "gauge"}, time)
+      d.emit({"metric" => "some.metric.name", "value" => 150.0, "type" => "counter"}, time)
+      d.emit({"metric" => "some.metric.name", "value" => 200.0, "type" => "counter"}, time)
+    end
+  end
+
+  it 'should be skipped if `metric` key does not exists' do
+    run_driver do |d, dog|
+      dog.should_receive(:emit_points).with(
+        "some.metric.name",
+        [[Time.parse("2014-02-08 04:14:15 UTC"), 50.0],
+         [Time.parse("2014-02-08 04:14:15 UTC"), 100.0]],
+        {"tags"=>["test.default"]}
+      )
+
+      log = d.instance.log
+      log.should_receive(:warn)
+         .with('`metric` key does not exist: {"no metric"=>"some.metric.name", "value"=>51.0, "time"=>"2014-02-08T04:14:15Z", "tag"=>"test.default"}')
+      log.should_receive(:warn)
+         .with('`metric` key does not exist: {"no metric"=>"some.metric.name", "value"=>101.0, "time"=>"2014-02-08T04:14:15Z", "tag"=>"test.default"}')
+
+      d.emit({"no metric" => "some.metric.name", "value" => 51.0}, time)
+      d.emit({"no metric" => "some.metric.name", "value" => 101.0}, time)
+      d.emit({"metric" => "some.metric.name", "value" => 50.0}, time)
+      d.emit({"metric" => "some.metric.name", "value" => 100.0}, time)
+    end
+  end
+end

spec/spec_helper.rb

@@ -0,0 +1,34 @@
+require 'fluent/test'
+require 'fluent/plugin/out_dd'
+require 'dogapi'
+require 'time'
+
+# Disable Test::Unit
+module Test::Unit::RunCount; def run(*); end; end
+
+RSpec.configure do |config|
+  config.before(:all) do
+    Fluent::Test.setup
+  end
+end
+
+def run_driver(options = {})
+  options = options.dup
+
+  dd_api_key = options[:dd_api_key] || 'test_dd_api_key'
+  host = options[:host] || 'test_host'
+
+  fluentd_conf = <<-EOS
+type datadog
+dd_api_key #{dd_api_key}
+host #{host}
+  EOS
+
+  tag = options[:tag] || 'test.default'
+  driver = Fluent::Test::OutputTestDriver.new(Fluent::DdOutput, tag).configure(fluentd_conf)
+
+  driver.run do
+    dog = driver.instance.instance_variable_get(:@dog)
+    yield(driver, dog)
+  end
+end