Skip to content

Use sync socket write to ensure that full message is written #39

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 1 commit into from
Closed

Use sync socket write to ensure that full message is written #39

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,6 @@
## 6.0.1
- Use sync socket write to ensure that full message is written to socket.

## 6.0.0
- Removed obsolete field `message_format`

Expand Down
9 changes: 1 addition & 8 deletions lib/logstash/outputs/tcp.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -151,14 +151,7 @@ def register
@codec.on_event do |event, payload|
begin
client_socket = connect unless client_socket
r,w,e = IO.select([client_socket], [client_socket], [client_socket], nil)
# don't expect any reads, but a readable socket might
# mean the remote end closed, so read it and throw it away.
# we'll get an EOFError if it happens.
client_socket.sysread(16384) if r.any?

# Now send the payload
client_socket.syswrite(payload) if w.any?
client_socket.syswrite(payload)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I believe the cause of truncated messages is likely caused by our failure to check the return value of IO#syswrite; if a payload is partially written at this level, we make no attempt to continue writing from where we left off.

We can fix that by performing a byteslice and repeating the IO#syswrite call with the remainder until the remaining payload is empty:

Suggested change
client_socket.syswrite(payload)
while payload && payload.bytesize > 0
written = client_socket.syswrite(payload)
payload = payload.byteslice(written..-1)
end

Separately, the code that this removes gives us an opportunity to have an EOFError be raised before we send data if the remote end closes the socket on us, while the replacement will raise a SystemCallError when IO#syswrite is sent to a closed socket. I have not thought through all of the ramifications of changing the error being raised.

I think there is definitely a bug in the implementation, in which a socket that is readable (e.g., the remote end has sent bytes that are not EOF) and not writable (e.g., has a zero window) would cause the payload to be dropped instead of written.

I believe this secondary issue can be solved by retrying the IO::select until we have a writable client_socket:

            begin
              r,w,_ = IO.select([client_socket],[client_socket])
              # don't expect any reads, but a readable socket might	
              # mean the remote end closed, so read it and throw it away.	
              # we'll get an EOFError if it happens.
              r.each { |readable| readable.sysread(16384) }
              retry unless w.any?
            end

There may be dragons here, if the socket never becomes writable or readable with an EOF, but these dragons also exist in the current implementation and are not a new addition.

rescue => e
@logger.warn("tcp output exception", :host => @host, :port => @port,
:exception => e, :backtrace => e.backtrace)
Expand Down
2 changes: 1 addition & 1 deletion logstash-output-tcp.gemspec
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
Gem::Specification.new do |s|

s.name = 'logstash-output-tcp'
s.version = '6.0.0'
s.version = '6.0.1'
s.licenses = ['Apache License (2.0)']
s.summary = "Writes events over a TCP socket"
s.description = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"
Expand Down