Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
F
fluent-plugin-wendelin
Project overview
Project overview
Details
Activity
Releases
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Issues
0
Issues
0
List
Boards
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Analytics
Analytics
CI / CD
Repository
Value Stream
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
Eteri
fluent-plugin-wendelin
Commits
ccf6ba26
Commit
ccf6ba26
authored
Jul 05, 2018
by
Eteri
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
use spaces instead of tabs
parent
3473c486
Changes
1
Hide whitespace changes
Inline
Side-by-side
Showing
1 changed file
with
153 additions
and
153 deletions
+153
-153
lib/fluent/plugin/wendelin_client.rb
lib/fluent/plugin/wendelin_client.rb
+153
-153
No files found.
lib/fluent/plugin/wendelin_client.rb
View file @
ccf6ba26
...
...
@@ -21,164 +21,164 @@ 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
#
@stop_timer = 0
end
# `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
#
@stop_timer = 0
end
# start request in an independent function to keep the connection open
def
start_request
(
uri
)
puts
" START NEW REQUEST"
@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
,)
end
# ingest `data_chunk` to a stream referenced as `reference`
def
ingest_with_keep_alive
(
reference
,
data_chunk
)
uri
=
URI
(
"
#{
@streamtool_uri
}
/ingest?reference=
#{
reference
}
"
)
def
start_request
(
uri
)
puts
" START NEW REQUEST"
@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
,)
end
# ingest `data_chunk` to a stream referenced as `reference`
def
ingest_with_keep_alive
(
reference
,
data_chunk
)
uri
=
URI
(
"
#{
@streamtool_uri
}
/ingest?reference=
#{
reference
}
"
)
puts
"uri = "
puts
uri
# 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
@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
.
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
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
.
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
# 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
@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
.
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
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
.
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
end
end
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment