diff --git a/lib/logstash/inputs/tcp.rb b/lib/logstash/inputs/tcp.rb index 897f46bc..bd96b706 100644 --- a/lib/logstash/inputs/tcp.rb +++ b/lib/logstash/inputs/tcp.rb @@ -181,7 +181,7 @@ def register validate_ssl_config! if server? - @loop = InputLoop.new(@id, @host, @port, DecoderImpl.new(@codec, self), @tcp_keep_alive, java_ssl_context) + @loop = InputLoop.new(@id, @host, @port, DecoderImpl.new(@codec, self), @tcp_keep_alive, java_ssl_context, @proxy_protocol) end end diff --git a/spec/inputs/tcp_spec.rb b/spec/inputs/tcp_spec.rb index 4aa04875..c7c2da3a 100644 --- a/spec/inputs/tcp_spec.rb +++ b/spec/inputs/tcp_spec.rb @@ -35,6 +35,23 @@ def get_port let(:port) { get_port } + def queue_pop_with_timeout(seconds, queue) + start = Time.now + deadline = start + seconds + item = nil + until item || Time.now >= deadline + begin + item = queue.pop(true) # raises ThreadError if empty + rescue ThreadError + sleep 0.1 + end + end + if item == nil + raise "Elapsed #{Time.now - start} seconds before pop a value" + end + return item + end + context "codec (PR #1372)" do it "switches from plain to line" do require "logstash/codecs/plain" @@ -108,18 +125,27 @@ def get_port } CONFIG - events = input(conf) do |pipeline, queue| + events = input(conf) do |_, queue| socket = Stud::try(5.times) { TCPSocket.new("127.0.0.1", port) } - socket.puts("PROXY TCP4 1.2.3.4 5.6.7.8 1234 5678\r"); + socket.write("PROXY TCP4 1.2.3.4 5.6.7.8 1234 5678\r") socket.flush - event_count.times do |i| - # unicode smiley for testing unicode support! - socket.puts("#{i} ☹") - socket.flush + # proceed with the rest of data writing in a separate thread + # so that crates some interleaving in the buffers seen on network layer + Thread.new do + sleep(1) + begin + socket.write("\n") + event_count.times do |i| + # unicode smiley for testing unicode support! + socket.puts("#{i} ☹") + socket.flush + end + socket.close + rescue => e + puts "Error while writing to the socket #{e.backtrace}" + end end - socket.close - - event_count.times.collect {queue.pop} + event_count.times.collect { queue_pop_with_timeout(5, queue) } end expect(events.length).to eq(event_count) diff --git a/src/main/java/org/logstash/tcp/InputLoop.java b/src/main/java/org/logstash/tcp/InputLoop.java index 86342d10..e2b914cf 100644 --- a/src/main/java/org/logstash/tcp/InputLoop.java +++ b/src/main/java/org/logstash/tcp/InputLoop.java @@ -67,9 +67,11 @@ public final class InputLoop implements Runnable, Closeable { * @param port Port to listen on * @param decoder {@link Decoder} provided by Jruby * @param keepAlive set to true to instruct the socket to issue TCP keep alive + * @param sslContext SSL configuration, or null when TLS is disabled + * @param proxy set to true to aggregate the HAProxy v1 header line before decoding */ public InputLoop(final String id, final String host, final int port, final Decoder decoder, final boolean keepAlive, - final SslContext sslContext) { + final SslContext sslContext, final boolean proxy) { this.sslContext = sslContext; this.host = host; this.port = port; @@ -79,7 +81,7 @@ public InputLoop(final String id, final String host, final int port, final Decod .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG, 1024) .childOption(ChannelOption.SO_KEEPALIVE, keepAlive) - .childHandler(new InputLoop.InputHandler(decoder, sslContext)); + .childHandler(new InputLoop.InputHandler(decoder, sslContext, proxy)); } @Override @@ -120,14 +122,16 @@ private static final class InputHandler extends ChannelInitializer out) throws Exception { + // Wait until we can decide: enough bytes to match the prefix, and if it is a PROXY + // line, the terminating \r\n must be present. + if (buffer.readableBytes() < PROXY_LENGTH) { + return; + } + if (startsWithProxy(buffer) && !containsCrlf(buffer)) { + return; + } + // Full PROXY line, or non-PROXY data: pass everything through and drop this handler + // so subsequent reads skip the aggregator entirely. + out.add(buffer.readRetainedSlice(buffer.readableBytes())); + ctx.pipeline().remove(this); + } + + private static boolean containsCrlf(ByteBuf buffer) { + int lfIndex = buffer.forEachByte(ByteProcessor.FIND_LF); + return lfIndex > 0 && buffer.getByte(lfIndex - 1) == '\r'; + } + + private static boolean startsWithProxy(ByteBuf buffer) { + for (int i = 0; i < PROXY_LENGTH; i++) { + if (buffer.getByte(buffer.readerIndex() + i) != PROXY_PREFIX[i]) { + return false; + } + } + return true; + } +} diff --git a/src/test/java/org/logstash/tcp/ProxyLineAggregatorTest.java b/src/test/java/org/logstash/tcp/ProxyLineAggregatorTest.java new file mode 100644 index 00000000..c7a8d172 --- /dev/null +++ b/src/test/java/org/logstash/tcp/ProxyLineAggregatorTest.java @@ -0,0 +1,65 @@ +package org.logstash.tcp; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.embedded.EmbeddedChannel; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.nio.charset.StandardCharsets; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.notNullValue; +import static org.hamcrest.Matchers.nullValue; + +class ProxyLineAggregatorTest { + + private EmbeddedChannel channel; + + @BeforeEach + void setUp() { + channel = new EmbeddedChannel(new ProxyLineAggregator()); + } + + @AfterEach + void tearDown() { + channel.finishAndReleaseAll(); + } + + private static ByteBuf ascii(String s) { + return Unpooled.copiedBuffer(s, StandardCharsets.US_ASCII); + } + + @Test + void partialProxyHeader_producesNoOutput() { + // "PRO" is shorter than the 5-byte "PROXY" prefix — decoder must wait + channel.writeInbound(ascii("PRO")); + assertThat(channel.readInbound(), nullValue()); + } + + @Test + void proxyPrefixWithoutCRLF_producesNoOutput() { + // Full PROXY keyword is present but the line terminator \r\n is missing + channel.writeInbound(ascii("PROXY TCP4 192.168.1.1 10.0.0.1 1234 80")); + assertThat(channel.readInbound(), nullValue()); + } + + @Test + void nonProxyDataWithSufficientBytes_passesThrough() { + // Content >= PROXY_LENGTH bytes but not a PROXY header and no \r\n: + // the aggregator should recognise it is not a PROXY line and let it through + channel.writeInbound(ascii("Hello, World!")); + ByteBuf result = channel.readInbound(); + assertThat(result, notNullValue()); + result.release(); + } + + @Test + void completeProxyLineWithCRLF_forwardsBuffer() { + channel.writeInbound(ascii("PROXY TCP4 192.168.1.1 10.0.0.1 1234 80\r\n")); + ByteBuf result = channel.readInbound(); + assertThat(result, notNullValue()); + result.release(); + } +}