fluent-plugin-wendelin: add keep connection open

lib/fluent/plugin/out_wendelin.rb

@@ -22,7 +22,6 @@
 require 'fluent/output'
 require_relative 'wendelin_client'
 
-
 module Fluent
 
   class WendelinOutput < ObjectBufferedOutput   # XXX verify base class
@@ -38,6 +37,8 @@ module Fluent
     config_param    :user,      :string,  :default => nil
     config_param    :password,  :string,  :default => nil
 
+    config_param    :reference,    :string
+
 
     def configure(conf)
         super
@@ -47,7 +48,7 @@ module Fluent
             credentials['user']     = @user
             credentials['password'] = @password
         end
-        @wendelin = WendelinClient.new(@streamtool_uri, credentials, @log)
+        @wendelin = WendelinClient.new(@streamtool_uri, credentials, @log, @reference)
     end
 
     def start
@@ -79,8 +80,8 @@ module Fluent
         # for input_stream_ref use tag as-is - it will be processed/translated
         # further on server by Wendelin
         reference = tag
+        @wendelin.ingest(data_chunk)
 
-        @wendelin.ingest(reference, data_chunk)
     end
 
   end

lib/fluent/plugin/wendelin_client.rb

@@ -19,64 +19,62 @@
 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)
+  def initialize(streamtool_uri, credentials, log, reference)
     @streamtool_uri = streamtool_uri
     @credentials    = credentials
     @log            = log
+    @reference = reference
+
   end
 
+  # start request in an independent function to keep the connection open
+  def start_request
+
+    uri = URI("#{@streamtool_uri}/ingest?reference=#{@reference}")
+    @http = Net::HTTP.start(uri.hostname, uri.port,
+      :use_ssl => (uri.scheme == 'https'),
+      :verify_mode  => OpenSSL::SSL::VERIFY_NONE,
+      :ssl_timeout  => 60,
+      :open_timeout => 60,
+      :read_timeout => 60,
+      :keep_alive_timeout => 60,)
+      @request = Net::HTTP::Post.new(uri)
+
+  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
+  def ingest(data_chunk)
 
-      # 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'
+    # 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 ||= start_request     # call start_request if request is undefined
+    @request.body = data_chunk
+    @request.content_type = 'application/octet-stream'
+    if @credentials.has_key?('user')
+      @request.basic_auth @credentials['user'], @credentials['password']
+    end
 
       @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 "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
-          # 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
+          res = @http.request(@request) # Net::HTTPResponse object
+      end
 
       rescue
           # some http/ssl/other connection error
@@ -100,6 +98,5 @@ class WendelinClient
               res.value
           end
       end
-  end
 
 end