fluent-plugin-wendelin: fix configuration parameters, check for closed connections

lib/fluent/plugin/out_wendelin.rb

@@ -22,6 +22,7 @@
 require 'fluent/output'
 require_relative 'wendelin_client'
 
+
 module Fluent
 
   class WendelinOutput < ObjectBufferedOutput   # XXX verify base class
@@ -37,8 +38,6 @@ module Fluent
     config_param    :user,      :string,  :default => nil
     config_param    :password,  :string,  :default => nil
 
-    config_param    :reference,    :string
-
 
     def configure(conf)
         super
@@ -48,7 +47,7 @@ module Fluent
             credentials['user']     = @user
             credentials['password'] = @password
         end
-        @wendelin = WendelinClient.new(@streamtool_uri, credentials, @log, @reference)
+        @wendelin = WendelinClient.new(@streamtool_uri, credentials, @log)
     end
 
     def start
@@ -80,8 +79,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,84 +19,97 @@
 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, reference)
-    @streamtool_uri = streamtool_uri
-    @credentials    = credentials
-    @log            = log
-    @reference = reference
+	# `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
+	end
 
-  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(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
-    @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=> #{@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.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 start_request(uri)
+
+		@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  => 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}")
+
+		# call start_request if request is undefined
+		@request ||= start_request(uri)
+
+		# connect again if the connection is not started
+		if ! @http.started?()
+			start_request(uri)
+		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
+		@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=> #{@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.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
 
 end