From 524b9dfa61ecfdc812a6e6861f90f76a7b9e75ac Mon Sep 17 00:00:00 2001 From: dimagolomozy Date: Tue, 3 Aug 2021 14:28:39 +0300 Subject: [PATCH] fix buffer index number for sticky tcp output connection --- output_tcp.go | 7 +++---- output_tcp_test.go | 9 +++++---- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/output_tcp.go b/output_tcp.go index f03e8ac..5be0d7c 100644 --- a/output_tcp.go +++ b/output_tcp.go @@ -95,16 +95,15 @@ func (o *TCPOutput) worker(bufferIndex int) { } } -func (o *TCPOutput) getBufferIndex(data []byte) int { +func (o *TCPOutput) getBufferIndex(msg *Message) int { if !o.config.Sticky { o.workerIndex++ return int(o.workerIndex) % o.config.Workers } hasher := fnv.New32a() - hasher.Write(payloadMeta(data)[1]) + hasher.Write(payloadID(msg.Meta)) return int(hasher.Sum32()) % o.config.Workers - } // PluginWrite writes message to this plugin @@ -113,7 +112,7 @@ func (o *TCPOutput) PluginWrite(msg *Message) (n int, err error) { return len(msg.Data), nil } - bufferIndex := o.getBufferIndex(msg.Data) + bufferIndex := o.getBufferIndex(msg) o.buf[bufferIndex] <- msg if Settings.OutputTCPStats { diff --git a/output_tcp_test.go b/output_tcp_test.go index 6026e8d..803601a 100644 --- a/output_tcp_test.go +++ b/output_tcp_test.go @@ -125,8 +125,9 @@ func TestBufferDistribution(t *testing.T) { } } -func getTestBytes() []byte { - reqh := payloadHeader(RequestPayload, uuid(), time.Now().UnixNano(), -1) - reqb := append(reqh, []byte("GET / HTTP/1.1\r\nHost: www.w3.org\r\nUser-Agent: Go 1.1 package http\r\nAccept-Encoding: gzip\r\n\r\n")...) - return reqb +func getTestBytes() *Message { + return &Message{ + Meta: payloadHeader(RequestPayload, uuid(), time.Now().UnixNano(), -1), + Data: []byte("GET / HTTP/1.1\r\nHost: www.w3.org\r\nUser-Agent: Go 1.1 package http\r\nAccept-Encoding: gzip\r\n\r\n"), + } }