0.14 support

README.md

@@ -85,13 +85,15 @@ The following configuration options are also supported:
   replace_invalid_utf8 false
 
   #buffered output options
-  retry_limit 40
-  retry_wait 5s
-  max_retry_wait 30s
-  flush_interval 5s
-  buffer_chunk_limit 100k
-  buffer_queue_limit 1024
-  num_threads 1
+  <buffer>
+    retry_max_times 40
+    retry_wait 5s
+    retry_max_interval 30s
+    flush_interval 5s
+    flush_thread_count 1
+    chunk_limit_size 100k
+    queue_limit_length 1024
+  </buffer>
 
 </match>
 ```
@@ -128,21 +130,21 @@ The cURL project maintains CA certificate bundles automatically converted from m
 
 ***replace_invalid_utf8*** - If this value is true and ***force_message_encoding*** is set to 'UTF-8' then all invalid UTF-8 sequences in log messages will be replaced with <?>.  Defaults to false.  This flag has no effect if ***force_message_encoding*** is not set to 'UTF-8'.
 
-####BufferedOutput options (inherited from Fluent::BufferedOutput)
+####Buffer options
 
-***retry_limit*** - the maximum number of times to retry a failed post request before giving up.  Defaults to *40*.
+***retry_max_times*** - the maximum number of times to retry a failed post request before giving up.  Defaults to *40*.
 
-***retry_wait*** - the initial time to wait before retrying a failed request.  Defaults to *5 seconds*.  Wait times will increase up to a maximum of ***max_retry_wait***
+***retry_wait*** - the initial time to wait before retrying a failed request.  Defaults to *5 seconds*.  Wait times will increase up to a maximum of ***retry_max_interval***
 
-***max_retry_wait*** - the maximum time to wait between retrying failed requests.  Defaults to *30 seconds*.  **Note:** This is not the total maximum time of all retry waits, but rather the maximum time to wait for a single retry.
+***retry_max_interval*** - the maximum time to wait between retrying failed requests.  Defaults to *30 seconds*.  **Note:** This is not the total maximum time of all retry waits, but rather the maximum time to wait for a single retry.
 
 ***flush_interval*** - how often to upload logs to Scalyr.  Defaults to *5 seconds*.
 
-***buffer_chunk_limit*** - the maximum amount of log data to send to Scalyr in a single request.  Defaults to *100KB*.  **Note:** if you set this value too large, then Scalyr may reject your requests.  Requests smaller than 1MB will typically be accepted by Scalyr, but note that the 1MB limit also includes the entire request body and all associated JSON keys and punctuation, which may be considerably larger than the raw log data.
+***flush_thread_count*** - the number of threads to use to upload logs.  This is currently fixed to 1 will cause fluentd to fail with a ConfigError if set to anything greater.
 
-***buffer_queue_limit*** - the maximum number of chunks to buffer before dropping new log requests.  Defaults to *1024*.  Combines with ***buffer_chunk_limit*** to give you the total amount of buffer to use in the event of request failures before dropping requests.
+***chunk_limit_size*** - the maximum amount of log data to send to Scalyr in a single request.  Defaults to *100KB*.  **Note:** if you set this value too large, then Scalyr may reject your requests.  Requests smaller than 1MB will typically be accepted by Scalyr, but note that the 1MB limit also includes the entire request body and all associated JSON keys and punctuation, which may be considerably larger than the raw log data.
 
-***num_threads*** - the number of threads to use to upload logs.  This is currently fixed to 1 will cause fluentd to fail with a ConfigError if set to anything greater.  
+***queue_limit_length*** - the maximum number of chunks to buffer before dropping new log requests.  Defaults to *1024*.  Combines with ***chunk_limit_size*** to give you the total amount of buffer to use in the event of request failures before dropping requests.
 
 Secondary Logging
 -----------------

VERSION

@@ -1 +1 @@
-0.7.4
+0.8.0

fluent-plugin-scalyr.gemspec

@@ -8,14 +8,14 @@ Gem::Specification.new do |gem|
   gem.version = File.read("VERSION").strip
   gem.authors = ["Imron Alston"]
   gem.licenses = ["Apache-2.0"]
-  gem.email = "imron@imralsoftware.com"
+  gem.email = "imron@scalyr.com"
   gem.has_rdoc = false
   gem.platform = Gem::Platform::RUBY
   gem.files = `git ls-files`.split("\n")
   gem.test_files = `git ls-files -- {test,spec,features}/*`.split("\n")
   gem.executables = `git ls-files -- bin/*`.split("\n").map{ |f| File.basename(f) }
   gem.require_paths = ['lib']
-  gem.add_dependency "fluentd", [">= 0.10.49", "< 0.14"]
+  gem.add_dependency "fluentd", [">= 0.14.0", "< 2"]
   gem.add_development_dependency "rake", "~> 0.9"
   gem.add_development_dependency "test-unit", "~> 3.0"
   gem.add_development_dependency "flexmock", "~> 1.2"

fluent.conf.sample

@@ -16,13 +16,14 @@
   # replace_invalid_utf8 false
 
 
-  ##BufferedOutput options
-  # retry_limit 40
-  # retry_wait 5s
-  # max_retry_wait 30s
-  # flush_interval 5s
-  # buffer_chunk_limit 100k
-  # buffer_queue_limit 1024
-  # num_threads 1
+  <buffer>
+  #  retry_max_times 40
+  #  retry_wait 5s
+  #  retry_max_interval 30s
+  #  flush_interval 5s
+  #  flush_thread_count 1
+  #  chunk_limit_size 100k
+  #  queue_limit_length 1024
+  </buffer>
 
 </match>

lib/fluent/plugin/out_scalyr.rb

@@ -16,16 +16,19 @@
 # limitations under the License.
 
 
+require 'fluent/plugin/output'
 require 'fluent/plugin/scalyr-exceptions'
+require 'fluent/plugin_helper/compat_parameters'
 require 'json'
 require 'net/http'
 require 'net/https'
 require 'securerandom'
 require 'thread'
 
 module Scalyr
-  class ScalyrOut < Fluent::BufferedOutput
+  class ScalyrOut < Fluent::Plugin::Output
     Fluent::Plugin.register_output( 'scalyr', self )
+    helpers :compat_parameters
 
     config_param :api_write_token, :string
     config_param :server_attributes, :hash, :default => nil
@@ -38,23 +41,35 @@ class ScalyrOut < Fluent::BufferedOutput
     config_param :force_message_encoding, :string, :default => nil
     config_param :replace_invalid_utf8, :bool, :default => false
 
-    config_set_default :retry_limit, 40 #try a maximum of 40 times before discarding
-    config_set_default :retry_wait, 5 #wait a minimum of 5 seconds before retrying again
-    config_set_default :max_retry_wait,  30 #wait a maximum of 30 seconds per retry
-    config_set_default :flush_interval, 5 #default flush interval of 5 seconds
+    config_section :buffer do
+      config_set_default :retry_max_times, 40 #try a maximum of 40 times before discarding
+      config_set_default :retry_max_interval,  30 #wait a maximum of 30 seconds per retry
+      config_set_default :retry_wait, 5 #wait a minimum of 5 seconds per retry
+      config_set_default :flush_interval, 5 #default flush interval of 5 seconds
+      config_set_default :chunk_limit_size, 1024*100 #default chunk size of 100k
+      config_set_default :queue_limit_length, 1024 #default queue size of 1024
+    end
+
+    # support for version 0.14.0:
+    def compat_parameters_default_chunk_key
+      ""
+    end
+
+    def formatted_to_msgpack_binary
+      true
+    end
 
     def configure( conf )
-      #need to call this before super because there doesn't seem to be any other way to
-      #set the default value for the buffer_chunk_limit, which is created and configured in super
-      if !conf.key? "buffer_chunk_limit"
-        conf["buffer_chunk_limit"] = "100k"
-      end
-      if !conf.key? "buffer_queue_limit"
-        conf["buffer_queue_limit"] = 1024
+
+      if conf.elements('buffer').empty?
+        $log.warn "Pre 0.14.0 configuration file detected.  Please consider updating your configuration file"
       end
+
+      compat_parameters_buffer( conf, default_chunk_key: '' )
+
       super
 
-      if @buffer.buffer_chunk_limit > 1024*1024
+      if @buffer.chunk_limit_size > 1024*1024
         $log.warn "Buffer chunk size is greater than 1Mb.  This may result in requests being rejected by Scalyr"
       end
 
@@ -76,8 +91,10 @@ def configure( conf )
 
       @add_events_uri = URI @scalyr_server + "addEvents"
 
+      num_threads = @buffer_config.flush_thread_count
+
       #forcibly limit the number of threads to 1 for now, to ensure requests always have incrementing timestamps
-      raise Fluent::ConfigError, "num_threads is currently limited to 1. You specified #{@num_threads}." if @num_threads > 1
+      raise Fluent::ConfigError, "num_threads is currently limited to 1. You specified #{num_threads}." if num_threads > 1
     end
 
     def start