change: plugin files [dev]

lib/fluent/plugin/out_wendelin.rb

@@ -19,14 +19,14 @@
 # This plugin hooks into Fluentd in a way similiar to out_forward and
 # out_secure_forward and outputs event stream to a Wendelin HTTP endpoint.
 
-require 'fluent/output'
+require 'fluent/plugin/output'
 require_relative 'wendelin_client'
 
 
-module Fluent
+module Fluent::Plugin
 
-  class WendelinOutput < ObjectBufferedOutput   # XXX verify base class
-    Plugin.register_output('wendelin', self)
+  class WendelinOutput < Output
+    Fluent::Plugin.register_output('wendelin', self)
 
     # where Wendelin's Input Stream Tool is located,
     # ex http://example.com/erp5/portal_ingestion_policies/example_ingestion
@@ -38,16 +38,32 @@ module Fluent
     config_param    :user,      :string,  :default => nil
     config_param    :password,  :string,  :default => nil
 
-
+    config_param    :use_keep_alive,  :bool,  :default => false
+
+    config_param    :ssl_timeout,        :integer,  :default => 60
+    config_param    :open_timeout,       :integer,  :default => 60
+    config_param    :read_timeout,       :integer,  :default => 60
+    config_param    :keep_alive_timeout, :integer,  :default => 300
+	
+	config_section :buffer do
+	  config_set_default :chunk_keys, ["tag"]
+	end
+	
     def configure(conf)
         super
+		
+		unless @chunk_key_tag
+		  raise Fluent::ConfigError, "buffer chunk key must include 'tag' for wendelin output"
+		end
 
         credentials = {}
         if @user
             credentials['user']     = @user
             credentials['password'] = @password
         end
-        @wendelin = WendelinClient.new(@streamtool_uri, credentials, @log)
+        @wendelin = WendelinClient.new(@streamtool_uri, credentials, @log,
+                                       @ssl_timeout, @open_timeout,
+                                       @read_timeout, @keep_alive_timeout)
     end
 
     def start
@@ -60,12 +76,11 @@ module Fluent
         # TODO
     end
 
-
-    # hooked to ObjectBufferedOutput - write out records from a buffer chunk for a tag.
+    # Use normal "Synchronous Buffer" - write out records from a buffer chunk for a tag.
     #
-    # NOTE this is called from a separate thread (see OutputThread and BufferedOutput)
-    def write_objects(tag, chunk)
-        # NOTE if this fail and raises -> it will unroll to BufferedOutput#try_flush
+    def write(chunk)
+	    return if chunk.empty?
+        # NOTE if this fail and raises -> it will unroll to Output#try_flush
         # which detects errors and retries outputting logs up to retry maxcount
         # times and aborts outputting current logs if all try fail.
         #
@@ -78,11 +93,16 @@ module Fluent
 
         # for input_stream_ref use tag as-is - it will be processed/translated
         # further on server by Wendelin
-        reference = tag
+        reference = chunk.metadata.tag
 
-        @wendelin.ingest(reference, data_chunk)
+        if @use_keep_alive
+          @wendelin.ingest_with_keep_alive(reference, data_chunk)
+        else
+          @wendelin.ingest(reference, data_chunk)
+        end
     end
 
   end
 
 end
+

lib/fluent/plugin/wendelin_client.rb

@@ -19,87 +19,171 @@
 require 'net/http'
 require 'openssl'
 
-
 # class representing a Wendelin client
 class WendelinClient
 
   # `streamtool_uri` - URI pointing to portal_input_data_stream "mountpoint"
   # `credentials`    # {'user' => _, 'password' => _}     TODO change to certificate
   # `log` - logger to use
-  def initialize(streamtool_uri, credentials, log)
-    @streamtool_uri = streamtool_uri
-    @credentials    = credentials
-    @log            = log
+  def initialize(streamtool_uri, credentials, log,
+                 ssl_timeout, open_timeout, read_timeout, keep_alive_timeout)
+    @streamtool_uri     = streamtool_uri
+    @credentials        = credentials
+    @log                = log
+    @ssl_timeout        = ssl_timeout
+    @open_timeout       = open_timeout
+    @read_timeout       = read_timeout
+    @keep_alive_timeout = keep_alive_timeout
   end
 
 
+  # start request in an independent function to keep the connection open
+  def start_connection(uri)
+
+    @log.debug "start new connection"
+
+    @http = Net::HTTP.start(uri.hostname, uri.port,
+      :use_ssl => (uri.scheme == 'https'),
+      :verify_mode  => OpenSSL::SSL::VERIFY_NONE,
+
+      # Net::HTTP default open timeout is infinity, which results
+      # in thread hang forever if other side does not fully
+      # establish connection. Default read_timeout is 60 seconds.
+      # We go safe way and make sure all timeouts are defined.
+      :ssl_timeout  => @ssl_timeout,
+      :open_timeout => @open_timeout,
+      :read_timeout => @read_timeout,
+      :keep_alive_timeout => @keep_alive_timeout,)
+  end
+
   # ingest `data_chunk` to a stream referenced as `reference`
-  def ingest(reference, data_chunk)
-      uri = URI("#{@streamtool_uri}/ingest?reference=#{reference}")
-      req = Net::HTTP::Post.new(uri)
-      if @credentials.has_key?('user')
-        req.basic_auth @credentials['user'], @credentials['password']
-      end
-
-      # TODO ensure content-type is 'raw', e.g. this way
-      # (but then querystring ?reference=... is lost)
-      # req.body = data_chunk
-      # req.content_type = 'application/octet-stream'
-      req.set_form_data('data_chunk' => data_chunk)
-
-      @log.on_trace do
-          @log.trace '>>> REQUEST'
-          @log.trace "method\t=> #{req.method}"
-          @log.trace "path\t=> #{req.path}"
-          @log.trace "uri\t=> #{req.uri}"
-          @log.trace "body\t=> #{req.body}"
-          @log.trace "body_stream\t=> #{req.body_stream}"
-          req.each {|h| @log.trace "#{h}:\t#{req[h]}"}
-          @log.trace
-      end
-
-      begin
-          # TODO keep connection open (so that every new ingest does not do
-          # full connect again)
-          res = Net::HTTP.start(uri.hostname, uri.port,
-                      :use_ssl      => (uri.scheme == 'https'),
-                      # NOTE = "do not check server cert"
-                      # TODO move this out to conf parameters
-                      :verify_mode  => OpenSSL::SSL::VERIFY_NONE,
-
-                      # Net::HTTP default open timeout is infinity, which results
-                      # in thread hang forever if other side does not fully
-                      # establish connection. Default read_timeout is 60 seconds.
-                      # We go safe way and make sure all timeouts are defined.
-                      :ssl_timeout  => 60,
-                      :open_timeout => 60,
-                      :read_timeout => 60,
-                ) do |http|
-              http.request(req)
-          end
-
-      rescue
-          # some http/ssl/other connection error
-          @log.warn "HTTP ERROR:"
-          raise
-
-      else
-          @log.on_trace do
-              @log.trace '>>> RESPONSE'
-              res.each {|h| @log.trace "#{h}:\t#{res[h]}"}
-              @log.trace "code\t=> #{res.code}"
-              @log.trace "msg\t=> #{res.message}"
-              @log.trace "class\t=> #{res.class}"
-              @log.trace "body:", res.body
-          end
-
-          if res.kind_of?(Net::HTTPSuccess)       # res.code is 2XX
-              #@log.info "ingested ok"
-          else
-              @log.warn "FAIL:"
-              res.value
-          end
-      end
+  def ingest_with_keep_alive(reference, data_chunk)
+    uri = URI("#{@streamtool_uri}/ingest?reference=#{reference}")
+    # call start_connection if http is undefined
+    if ! defined? @http
+      start_connection(uri)
+    end
+
+    # connect again if the connection is not started
+    if ! @http.started?()
+      start_connection(uri)
+    end
+
+    @request = Net::HTTP::Post.new(uri)
+
+    # When using 'application/x-www-form-urlencoded', Ruby encodes with regex
+    # and it is far too slow. Such POST is legit:
+    # https://stackoverflow.com/a/14710450
+    @request.body = data_chunk
+    @request.content_type = 'application/octet-stream'
+
+    if @credentials.has_key?('user')
+      @request.basic_auth @credentials['user'], @credentials['password']
+    end
+
+    @log.trace do
+      @log.trace '>>> REQUEST'
+      @log.trace "method\t=> #{@request.method}"
+      @log.trace "path\t=> #{@request.path}"
+      @log.trace "uri\t=> #{@request.uri}"
+      @log.trace "body\t=> #{@request.body}"
+      @log.trace "body_stream\t=> #{@request.body_stream}"
+      @request.each {|h| @log.trace "#{h}:\t#{@request[h]}"}
+      @log.trace
+    end
+
+    begin
+      res = @http.request(@request) # Net::HTTPResponse object
+    end
+
+    rescue
+      # some http/ssl/other connection error
+      @log.warn "HTTP ERROR:"
+    raise
+  else
+    @log.trace do
+      @log.trace '>>> RESPONSE'
+      res.each {|h| @log.trace "#{h}:\t#{res[h]}"}
+      @log.trace "code\t=> #{res.code}"
+      @log.trace "msg\t=> #{res.message}"
+      @log.trace "class\t=> #{res.class}"
+      @log.trace "body:", res.body
+    end
+
+    if res.kind_of?(Net::HTTPSuccess)       # res.code is 2XX
+      #@log.info "ingested ok"
+    else
+      @log.warn "FAIL:"
+      res.value
+    end
   end
 
+  def ingest(reference, data_chunk)
+     uri = URI("#{@streamtool_uri}/ingest?reference=#{reference}")
+     req = Net::HTTP::Post.new(uri)
+     if @credentials.has_key?('user')
+       req.basic_auth @credentials['user'], @credentials['password']
+     end
+
+     # When using 'application/x-www-form-urlencoded', Ruby encodes with regex
+     # and it is far too slow. Such POST is legit:
+     # https://stackoverflow.com/a/14710450
+     req.body = data_chunk
+     req.content_type = 'application/octet-stream'
+
+     @log.trace do
+         @log.trace '>>> REQUEST'
+         @log.trace "method\t=> #{req.method}"
+         @log.trace "path\t=> #{req.path}"
+         @log.trace "uri\t=> #{req.uri}"
+         @log.trace "body\t=> #{req.body}"
+         @log.trace "body_stream\t=> #{req.body_stream}"
+         req.each {|h| @log.trace "#{h}:\t#{req[h]}"}
+         @log.trace
+     end
+
+     begin
+         # TODO keep connection open (so that every new ingest does not do
+         # full connect again)
+         res = Net::HTTP.start(uri.hostname, uri.port,
+                     :use_ssl      => (uri.scheme == 'https'),
+                     # NOTE = "do not check server cert"
+                     # TODO move this out to conf parameters
+                     :verify_mode  => OpenSSL::SSL::VERIFY_NONE,
+
+                     # Net::HTTP default open timeout is infinity, which results
+                     # in thread hang forever if other side does not fully
+                     # establish connection. Default read_timeout is 60 seconds.
+                     # We go safe way and make sure all timeouts are defined.
+                     :ssl_timeout  => @ssl_timeout,
+                     :open_timeout => @open_timeout,
+                     :read_timeout => @read_timeout,
+               ) do |http|
+             http.request(req)
+         end
+
+     rescue
+         # some http/ssl/other connection error
+         @log.warn "HTTP ERROR:"
+         raise
+
+     else
+         @log.trace do
+             @log.trace '>>> RESPONSE'
+             res.each {|h| @log.trace "#{h}:\t#{res[h]}"}
+             @log.trace "code\t=> #{res.code}"
+             @log.trace "msg\t=> #{res.message}"
+             @log.trace "class\t=> #{res.class}"
+             @log.trace "body:", res.body
+         end
+
+         if res.kind_of?(Net::HTTPSuccess)       # res.code is 2XX
+             #@log.info "ingested ok"
+         else
+             @log.warn "FAIL:"
+             res.value
+         end
+     end
+ end
+
 end