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
f1c7bcc4
Commit
f1c7bcc4
authored
Apr 20, 2018
by
Eteri
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
solve different tag problem
parent
a8b179a1
Changes
1
Hide whitespace changes
Inline
Side-by-side
Showing
1 changed file
with
84 additions
and
88 deletions
+84
-88
lib/fluent/plugin/wendelin_client.rb
lib/fluent/plugin/wendelin_client.rb
+84
-88
No files found.
lib/fluent/plugin/wendelin_client.rb
View file @
f1c7bcc4
...
...
@@ -15,99 +15,95 @@
# See the License for the specific language governing permissions and
# limitations under the License.
require
'net/http'
require
'openssl'
require
'http'
# 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
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
end
# start request in an independent function to keep the connection open
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
,)
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
)
@request
=
Net
::
HTTP
::
Post
.
new
(
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
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
(
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
start_request
(
uri
)
unless
@http
.
started?
@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
.
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
StandardError
# 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
.
is_a?
(
Net
::
HTTPSuccess
)
# res.code is 2XX
# @log.info "ingested ok"
else
@log
.
warn
'FAIL:'
res
.
value
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