Merge pull request #981 from DimaGolomozy/fix-sticky-buffer-index

fix buffer index number for sticky tcp output connection

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 {

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"),
+	}
 }