Merge pull request fakechris#2 from johanneswuerbach/batch_byte_size

Configurable batch_byte_size

Gemfile.lock

@@ -1,30 +1,29 @@
 PATH
   remote: .
   specs:
-    fluent-plugin-statsd (1.1.0)
+    fluent-plugin-statsd-output (1.1.1)
       fluentd (>= 0.10.8)
+      statsd-ruby (~> 1.4)
 
 GEM
   remote: http://rubygems.org/
   specs:
     coderay (1.1.1)
-    cool.io (1.4.4)
+    cool.io (1.5.0)
     diff-lcs (1.2.5)
-    fluentd (0.14.1)
-      cool.io (>= 1.4.3, < 2.0.0)
+    fluentd (0.14.17)
+      cool.io (>= 1.4.5, < 2.0.0)
       http_parser.rb (>= 0.5.1, < 0.7.0)
-      json (>= 1.4.3)
-      msgpack (>= 0.7.0)
-      serverengine (>= 1.6.4)
+      msgpack (>= 0.7.0, < 2.0.0)
+      serverengine (>= 2.0.4, < 3.0.0)
       sigdump (~> 0.2.2)
-      strptime (>= 0.1.7)
-      tzinfo (>= 1.0.0)
-      tzinfo-data (>= 1.0.0)
+      strptime (~> 0.1.7)
+      tzinfo (~> 1.0)
+      tzinfo-data (~> 1.0)
       yajl-ruby (~> 1.0)
     http_parser.rb (0.6.0)
-    json (2.0.1)
     method_source (0.8.2)
-    msgpack (1.0.0)
+    msgpack (1.1.0)
     power_assert (0.3.0)
     pry (0.10.4)
       coderay (~> 1.1.0)
@@ -43,30 +42,30 @@ GEM
       diff-lcs (>= 1.2.0, < 2.0)
       rspec-support (~> 3.5.0)
     rspec-support (3.5.0)
-    serverengine (1.6.4)
+    serverengine (2.0.5)
       sigdump (~> 0.2.2)
     sigdump (0.2.4)
     slop (3.6.0)
-    statsd-ruby (1.3.0)
-    strptime (0.1.8)
+    statsd-ruby (1.4.0)
+    strptime (0.1.9)
     test-unit (3.1.8)
       power_assert
-    thread_safe (0.3.5)
-    tzinfo (1.2.2)
+    thread_safe (0.3.6)
+    tzinfo (1.2.3)
       thread_safe (~> 0.1)
-    tzinfo-data (1.2016.6)
+    tzinfo-data (1.2017.2)
       tzinfo (>= 1.0.0)
-    yajl-ruby (1.2.1)
+    yajl-ruby (1.3.0)
 
 PLATFORMS
   ruby
 
 DEPENDENCIES
-  fluent-plugin-statsd!
+  fluent-plugin-statsd-output!
   pry
   rspec (~> 3.0)
   statsd-ruby
   test-unit
 
 BUNDLED WITH
-   1.10.6
+   1.13.6

README.md

@@ -16,6 +16,7 @@ $ fluent-gem install fluent-plugin-statsd-output
   host localhost # optional
   port 8125 # optional
   namespace a.b.c # optional
+  batch_byte_size 512 # optional
 
   <metric>
     statsd_type timing

fluent-plugin-statsd.gemspec

@@ -16,7 +16,7 @@ Gem::Specification.new do |gem|
   gem.require_paths = ['lib']
 
   gem.add_dependency "fluentd", ">= 0.10.8"
-  gem.add_dependency "statsd-ruby", "~> 1.3.0"
+  gem.add_dependency "statsd-ruby", "~> 1.4"
 
   gem.add_development_dependency "rspec", '~> 3.0'
   gem.add_development_dependency "test-unit"

lib/fluent/plugin/out_statsd.rb

@@ -10,6 +10,7 @@ class StatsdOutput < BufferedOutput
     config_param :host, :string, :default => 'localhost'
     config_param :port, :string, :default => '8125'
     config_param :namespace, :string, :default => nil
+    config_param :batch_byte_size, :integer, :default => nil
 
     config_section :metric do
       config_param :statsd_type, :string
@@ -27,6 +28,11 @@ def configure(conf)
       super
       @statsd = Statsd.new(host, port)
       @statsd.namespace = namespace if namespace
+
+      if batch_byte_size
+        @statsd.batch_size = nil
+        @statsd.batch_byte_size = batch_byte_size
+      end
       log.info(statsd)
 
       @metrics = conf.elements.select {|elem| elem.name == 'metric' }

spec/plugin/out_statsd_spec.rb

@@ -6,6 +6,7 @@
     %{
       type statsd
       namespace a.b.c
+      batch_byte_size 512
 
       <metric>
         statsd_type timing
@@ -20,7 +21,13 @@
     }
   end
   let(:driver) { create_driver(config) }
-  let(:statsd) { double('statsd', increment: true, timing: true, 'namespace=' => true) }
+  let(:statsd) { double('statsd', increment: true,
+                                  timing: true,
+                                  'namespace=' => true,
+                                  'batch_size=' => true,
+                                  'batch_byte_size' => true)
+
+               }
   let(:time) { Time.now.utc }
 
   before :all do
@@ -41,6 +48,8 @@ def emit_events(events)
     allow(Statsd).to receive(:new).and_return(statsd)
 
     expect(statsd).to receive(:namespace=).with('a.b.c')
+    expect(statsd).to receive(:batch_size=).with(nil)
+    expect(statsd).to receive(:batch_byte_size=).with(512)
 
     expect(statsd).to receive(:increment).with('res_code_2xx').twice.times
     expect(statsd).to receive(:increment).with('res_code_4xx').once.times