Commit 72f1a151 authored by Eteri's avatar Eteri Committed by Klaus Wölfel

fluent-plugin-wendelin: add keep connection open

parent f0b1dcca
...@@ -22,7 +22,6 @@ ...@@ -22,7 +22,6 @@
require 'fluent/output' require 'fluent/output'
require_relative 'wendelin_client' require_relative 'wendelin_client'
module Fluent module Fluent
class WendelinOutput < ObjectBufferedOutput # XXX verify base class class WendelinOutput < ObjectBufferedOutput # XXX verify base class
...@@ -38,6 +37,8 @@ module Fluent ...@@ -38,6 +37,8 @@ module Fluent
config_param :user, :string, :default => nil config_param :user, :string, :default => nil
config_param :password, :string, :default => nil config_param :password, :string, :default => nil
config_param :reference, :string
def configure(conf) def configure(conf)
super super
...@@ -47,7 +48,7 @@ module Fluent ...@@ -47,7 +48,7 @@ module Fluent
credentials['user'] = @user credentials['user'] = @user
credentials['password'] = @password credentials['password'] = @password
end end
@wendelin = WendelinClient.new(@streamtool_uri, credentials, @log) @wendelin = WendelinClient.new(@streamtool_uri, credentials, @log, @reference)
end end
def start def start
...@@ -79,8 +80,8 @@ module Fluent ...@@ -79,8 +80,8 @@ module Fluent
# for input_stream_ref use tag as-is - it will be processed/translated # for input_stream_ref use tag as-is - it will be processed/translated
# further on server by Wendelin # further on server by Wendelin
reference = tag reference = tag
@wendelin.ingest(data_chunk)
@wendelin.ingest(reference, data_chunk)
end end
end end
......
...@@ -19,63 +19,61 @@ ...@@ -19,63 +19,61 @@
require 'net/http' require 'net/http'
require 'openssl' require 'openssl'
# class representing a Wendelin client # class representing a Wendelin client
class WendelinClient class WendelinClient
# `streamtool_uri` - URI pointing to portal_input_data_stream "mountpoint" # `streamtool_uri` - URI pointing to portal_input_data_stream "mountpoint"
# `credentials` # {'user' => _, 'password' => _} TODO change to certificate # `credentials` # {'user' => _, 'password' => _} TODO change to certificate
# `log` - logger to use # `log` - logger to use
def initialize(streamtool_uri, credentials, log) def initialize(streamtool_uri, credentials, log, reference)
@streamtool_uri = streamtool_uri @streamtool_uri = streamtool_uri
@credentials = credentials @credentials = credentials
@log = log @log = log
@reference = reference
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)
# 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 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 # When using 'application/x-www-form-urlencoded', Ruby encodes with regex
# and it is far too slow. Such POST is legit: # and it is far too slow. Such POST is legit:
# https://stackoverflow.com/a/14710450 # https://stackoverflow.com/a/14710450
req.body = data_chunk @request ||= start_request # call start_request if request is undefined
req.content_type = 'application/octet-stream' @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.on_trace do
@log.trace '>>> REQUEST' @log.trace '>>> REQUEST'
@log.trace "method\t=> #{req.method}" @log.trace "method\t=> #{@request.method}"
@log.trace "path\t=> #{req.path}" @log.trace "path\t=> #{@request.path}"
@log.trace "uri\t=> #{req.uri}" @log.trace "uri\t=> #{@request.uri}"
@log.trace "body\t=> #{req.body}" @log.trace "body\t=> #{@request.body}"
@log.trace "body_stream\t=> #{req.body_stream}" @log.trace "body_stream\t=> #{@request.body_stream}"
req.each {|h| @log.trace "#{h}:\t#{req[h]}"} @request.each {|h| @log.trace "#{h}:\t#{@request[h]}"}
@log.trace @log.trace
end end
begin begin
# TODO keep connection open (so that every new ingest does not do res = @http.request(@request) # Net::HTTPResponse object
# 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 end
rescue rescue
...@@ -100,6 +98,5 @@ class WendelinClient ...@@ -100,6 +98,5 @@ class WendelinClient
res.value res.value
end end
end end
end
end end
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment