From c5d1112e7ab3147fd985e9cafc855d75bf585755 Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Sun, 27 May 2018 12:31:47 +0300 Subject: [PATCH] Add way to control packet capture buffer size and optimize snaplen MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Added `—input-raw-buffer-size` - Controls size of the OS buffer (in bytes) which holds packets until they dispatched. Default value depends by system: in Linux around 2MB. If you see big package drop, increase this value. Additionally snaplen (max number of bytes being read for each packet) now dynamically set based on interface MTU + max header size. In most situations it should reduce package drop, because each packet will consume less space in buffer. --- elasticsearch_test.go | 2 +- http_client.go | 4 ++-- input_kafka.go | 2 +- input_raw.go | 6 ++++-- input_raw_test.go | 14 ++++++------- limiter.go | 2 +- middleware_test.go | 4 ++-- plugins.go | 2 +- raw_socket_listener/listener.go | 18 +++++++++++++++-- raw_socket_listener/listener_test.go | 30 ++++++++++++++-------------- raw_socket_listener/tcp_message.go | 6 +++--- settings.go | 3 +++ vendor/vendor.json | 6 +++--- 13 files changed, 59 insertions(+), 40 deletions(-) diff --git a/elasticsearch_test.go b/elasticsearch_test.go index 27e624b..7bddcff 100644 --- a/elasticsearch_test.go +++ b/elasticsearch_test.go @@ -28,7 +28,7 @@ func assertExpectedError(returnedError error, t *testing.T) { func assertNoError(returnedError error, t *testing.T) { if nil != returnedError { - t.Errorf("Expected err %s but got %s", nil, returnedError) + t.Errorf("Expected no err but got %s", returnedError) } } diff --git a/http_client.go b/http_client.go index b6c27b6..02285fc 100644 --- a/http_client.go +++ b/http_client.go @@ -113,7 +113,7 @@ func (c *HTTPClient) Disconnect() { } } -func (c *HTTPClient) isAlive(readBytes *int) (bool) { +func (c *HTTPClient) isAlive(readBytes *int) bool { // Ready 1 byte from socket without timeout to check if it not closed c.conn.SetReadDeadline(time.Now().Add(time.Millisecond)) n, err := c.conn.Read(c.respBuf[:1]) @@ -203,7 +203,7 @@ func (c *HTTPClient) Send(data []byte) (response []byte, err error) { currentContentLength += n } else { // If headers are finished - var firstEmptyLine = bytes.Index(c.respBuf[:readBytes], proto.EmptyLine); + var firstEmptyLine = bytes.Index(c.respBuf[:readBytes], proto.EmptyLine) if firstEmptyLine != -1 { if bytes.Equal(proto.Header(c.respBuf[:readBytes], []byte("Transfer-Encoding")), []byte("chunked")) { chunked = true diff --git a/input_kafka.go b/input_kafka.go index 343cd11..90c17cd 100644 --- a/input_kafka.go +++ b/input_kafka.go @@ -29,7 +29,7 @@ func NewKafkaInput(address string, config *KafkaConfig) *KafkaInput { } else { var err error //con, err = sarama.NewConsumer([]string{config.host}, c) - con, err = sarama.NewConsumer(strings.Split(config.host,","), c) + con, err = sarama.NewConsumer(strings.Split(config.host, ","), c) if err != nil { log.Fatalln("Failed to start Sarama(Kafka) consumer:", err) diff --git a/input_raw.go b/input_raw.go index 7dddce1..1cd49c6 100644 --- a/input_raw.go +++ b/input_raw.go @@ -21,6 +21,7 @@ type RAWInput struct { listener *raw.Listener bpfFilter string timestampType string + bufferSize int } // Available engines for intercepting traffic @@ -31,7 +32,7 @@ const ( ) // NewRAWInput constructor for RAWInput. Accepts address with port as argument. -func NewRAWInput(address string, engine int, trackResponse bool, expire time.Duration, realIPHeader string, bpfFilter string, timestampType string) (i *RAWInput) { +func NewRAWInput(address string, engine int, trackResponse bool, expire time.Duration, realIPHeader string, bpfFilter string, timestampType string, bufferSize int) (i *RAWInput) { i = new(RAWInput) i.data = make(chan *raw.TCPMessage) i.address = address @@ -42,6 +43,7 @@ func NewRAWInput(address string, engine int, trackResponse bool, expire time.Dur i.quit = make(chan bool) i.trackResponse = trackResponse i.timestampType = timestampType + i.bufferSize = bufferSize i.listen(address) i.listener.IsReady() @@ -79,7 +81,7 @@ func (i *RAWInput) listen(address string) { log.Fatal("input-raw: error while parsing address", err) } - i.listener = raw.NewListener(host, port, i.engine, i.trackResponse, i.expire, i.bpfFilter, i.timestampType) + i.listener = raw.NewListener(host, port, i.engine, i.trackResponse, i.expire, i.bpfFilter, i.timestampType, i.bufferSize) ch := i.listener.Receiver() diff --git a/input_raw_test.go b/input_raw_test.go index cd0b319..517bdbd 100644 --- a/input_raw_test.go +++ b/input_raw_test.go @@ -44,7 +44,7 @@ func TestRAWInputIPv4(t *testing.T) { var respCounter, reqCounter int64 - input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "X-Real-IP", "", "") + input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "X-Real-IP", "", "", 0) defer input.Close() output := NewTestOutput(func(data []byte) { @@ -106,7 +106,7 @@ func TestRAWInputNoKeepAlive(t *testing.T) { originAddr := listener.Addr().String() - input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "", "", "") + input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "", "", "", 0) defer input.Close() output := NewTestOutput(func(data []byte) { @@ -152,7 +152,7 @@ func TestRAWInputIPv6(t *testing.T) { var respCounter, reqCounter int64 - input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "", "", "") + input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "", "", "", 0) defer input.Close() output := NewTestOutput(func(data []byte) { @@ -203,7 +203,7 @@ func TestInputRAW100Expect(t *testing.T) { originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1) - input := NewRAWInput(originAddr, EnginePcap, true, time.Second, "", "", "") + input := NewRAWInput(originAddr, EnginePcap, true, time.Second, "", "", "", 0) defer input.Close() // We will use it to get content of raw HTTP request @@ -266,7 +266,7 @@ func TestInputRAWChunkedEncoding(t *testing.T) { })) originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1) - input := NewRAWInput(originAddr, EnginePcap, true, time.Second, "", "", "") + input := NewRAWInput(originAddr, EnginePcap, true, time.Second, "", "", "", 0) defer input.Close() replay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -330,7 +330,7 @@ func TestInputRAWLargePayload(t *testing.T) { })) originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1) - input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "", "", "") + input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "", "", "", 0) defer input.Close() replay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { @@ -375,7 +375,7 @@ func BenchmarkRAWInput(b *testing.B) { var respCounter, reqCounter int64 - input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "", "", "") + input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "", "", "", 0) defer input.Close() output := NewTestOutput(func(data []byte) { diff --git a/limiter.go b/limiter.go index cb1a36c..c28ff62 100644 --- a/limiter.go +++ b/limiter.go @@ -96,5 +96,5 @@ func (l *Limiter) Read(data []byte) (n int, err error) { } func (l *Limiter) String() string { - return fmt.Sprintf("Limiting %s to: %d (isPercent: %b)", l.plugin, l.limit, l.isPercent) + return fmt.Sprintf("Limiting %s to: %d (isPercent: %v)", l.plugin, l.limit, l.isPercent) } diff --git a/middleware_test.go b/middleware_test.go index b8d3c9a..28f65dc 100644 --- a/middleware_test.go +++ b/middleware_test.go @@ -118,7 +118,7 @@ func TestEchoMiddleware(t *testing.T) { // Catch traffic from one service fromAddr := strings.Replace(from.Listener.Addr().String(), "[::]", "127.0.0.1", -1) - input := NewRAWInput(fromAddr, EnginePcap, true, testRawExpire, "", "", "") + input := NewRAWInput(fromAddr, EnginePcap, true, testRawExpire, "", "", "", 0) defer input.Close() // And redirect to another @@ -180,7 +180,7 @@ func TestTokenMiddleware(t *testing.T) { fromAddr := strings.Replace(from.Listener.Addr().String(), "[::]", "127.0.0.1", -1) // Catch traffic from one service - input := NewRAWInput(fromAddr, EnginePcap, true, testRawExpire, "", "", "") + input := NewRAWInput(fromAddr, EnginePcap, true, testRawExpire, "", "", "", 0) defer input.Close() // And redirect to another diff --git a/plugins.go b/plugins.go index 1cb0637..7cc2f7a 100644 --- a/plugins.go +++ b/plugins.go @@ -106,7 +106,7 @@ func InitPlugins() { } for _, options := range Settings.inputRAW { - registerPlugin(NewRAWInput, options, engine, Settings.inputRAWTrackResponse, Settings.inputRAWExpire, Settings.inputRAWRealIPHeader, Settings.inputRAWBpfFilter, Settings.inputRAWTimestampType) + registerPlugin(NewRAWInput, options, engine, Settings.inputRAWTrackResponse, Settings.inputRAWExpire, Settings.inputRAWRealIPHeader, Settings.inputRAWBpfFilter, Settings.inputRAWTimestampType, Settings.inputRawBufferSize) } for _, options := range Settings.inputTCP { diff --git a/raw_socket_listener/listener.go b/raw_socket_listener/listener.go index ad1e373..f4d7246 100644 --- a/raw_socket_listener/listener.go +++ b/raw_socket_listener/listener.go @@ -75,6 +75,8 @@ type Listener struct { bpfFilter string timestampType string + bufferSize int + conn net.PacketConn pcapHandles []*pcap.Handle @@ -96,7 +98,7 @@ const ( ) // NewListener creates and initializes new Listener object -func NewListener(addr string, port string, engine int, trackResponse bool, expire time.Duration, bpfFilter string, timestampType string) (l *Listener) { +func NewListener(addr string, port string, engine int, trackResponse bool, expire time.Duration, bpfFilter string, timestampType string, bufferSize int) (l *Listener) { l = &Listener{} l.packetsChan = make(chan *packet, 10000) @@ -112,6 +114,7 @@ func NewListener(addr string, port string, engine int, trackResponse bool, expir l.trackResponse = trackResponse l.bpfFilter = bpfFilter l.timestampType = timestampType + l.bufferSize = bufferSize l.addr = addr _port, _ := strconv.Atoi(port) @@ -345,10 +348,21 @@ func (t *Listener) readPcap() { log.Println("Supported timestamp types: ", inactive.SupportedTimestamps(), device.Name) } } - inactive.SetSnapLen(65536) + + if it, err := net.InterfaceByName(device.Name); err == nil { + // Auto-guess max length of packet to capture + inactive.SetSnapLen(it.MTU + 68*2) + } else { + inactive.SetSnapLen(65536) + } + inactive.SetTimeout(t.messageExpire) inactive.SetPromisc(true) + if t.bufferSize > 0 { + inactive.SetBufferSize(t.bufferSize) + } + handle, herr := inactive.Activate() if herr != nil { log.Println("PCAP Activate error:", herr) diff --git a/raw_socket_listener/listener_test.go b/raw_socket_listener/listener_test.go index a9bf502..2a6d9c0 100644 --- a/raw_socket_listener/listener_test.go +++ b/raw_socket_listener/listener_test.go @@ -12,7 +12,7 @@ import ( func TestRawListenerInput(t *testing.T) { var req, resp *TCPMessage - listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "", "") + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "", "", 0) defer listener.Close() reqPacket := buildPacket(true, 1, 1, []byte("GET / HTTP/1.1\r\n\r\n"), time.Now()) @@ -77,7 +77,7 @@ func responsePacket(prev *TCPPacket, payload []byte) *TCPPacket { } func TestHEADRequestNoBody(t *testing.T) { - listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "") + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "", "", 0) defer listener.Close() reqPacket := firstPacket([]byte("HEAD / HTTP/1.1\r\nContent-Length: 0\r\n\r\n")) @@ -89,7 +89,7 @@ func TestHEADRequestNoBody(t *testing.T) { var req, resp *TCPMessage select { case req = <-listener.messagesChan: - case <-time.After( time.Millisecond): + case <-time.After(time.Millisecond): t.Error("Should return request immediately") return } @@ -111,7 +111,7 @@ func TestHEADRequestNoBody(t *testing.T) { } func TestSingleAck100Continue(t *testing.T) { - listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "", "") + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "", "", 0) defer listener.Close() reqPacket1 := firstPacket([]byte("POST / HTTP/1.1\r\nExpect: 100-continue\r\nContent-Length: 4\r\n\r\n")) @@ -130,7 +130,7 @@ func TestSingleAck100Continue(t *testing.T) { } func Test100ContinueWithoutWaiting(t *testing.T) { - listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "", "") + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "", "", 0) defer listener.Close() req1 := firstPacket([]byte("POST / HTTP/1.1\r\nExpect: 100-continue\r\nContent-Length: 4\r\n\r\n")) @@ -146,7 +146,7 @@ func Test100ContinueWithoutWaiting(t *testing.T) { // Client first sends data without waiting 100-continue, but once response received, generate packets based on Ack payload func Test100ContinueMixed(t *testing.T) { - listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "", "") + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "", "", 0) defer listener.Close() req1 := firstPacket([]byte("POST / HTTP/1.1\r\nExpect: 100-continue\r\nContent-Length: 12\r\n\r\n")) @@ -164,7 +164,7 @@ func Test100ContinueMixed(t *testing.T) { } func TestDoubleAck100Continue(t *testing.T) { - listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "", "") + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "", "", 0) defer listener.Close() reqPacket1 := firstPacket([]byte("POST / HTTP/1.1\r\nExpect: 100-continue\r\nContent-Length: 4\r\n\r\n")) @@ -187,7 +187,7 @@ func TestDoubleAck100Continue(t *testing.T) { func TestRawListenerInputResponseByClose(t *testing.T) { var req, resp *TCPMessage - listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "", "") + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "", "", 0) defer listener.Close() reqPacket := buildPacket(true, 1, 1, []byte("GET / HTTP/1.1\r\n\r\n"), time.Now()) @@ -227,7 +227,7 @@ func TestRawListenerInputResponseByClose(t *testing.T) { func TestRawListenerInputWithoutResponse(t *testing.T) { var req *TCPMessage - listener := NewListener("", "0", EnginePcap, false, 10*time.Millisecond, "", "") + listener := NewListener("", "0", EnginePcap, false, 10*time.Millisecond, "", "", 0) defer listener.Close() reqPacket := buildPacket(true, 1, 1, []byte("GET / HTTP/1.1\r\n\r\n"), time.Now()) @@ -249,7 +249,7 @@ func TestRawListenerInputWithoutResponse(t *testing.T) { func TestRawListenerResponse(t *testing.T) { var req, resp *TCPMessage - listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "", "") + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "", "", 0) defer listener.Close() reqPacket := firstPacket([]byte("GET / HTTP/1.1\r\n\r\n")) @@ -297,7 +297,7 @@ func get100ContinuePackets() (req []*TCPPacket, resp []*TCPPacket) { } func TestShort100Continue(t *testing.T) { - listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "", "") + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "", "", 0) defer listener.Close() req, resp := get100ContinuePackets() @@ -309,7 +309,7 @@ func TestShort100Continue(t *testing.T) { // Response comes before Request func Test100ContinueWrongOrder(t *testing.T) { - listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "", "") + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "", "", 0) defer listener.Close() req, resp := get100ContinuePackets() @@ -462,7 +462,7 @@ func permutation(n int, list []*TCPPacket) []*TCPPacket { // Response comes before Request func TestRawListenerChunkedWrongOrder(t *testing.T) { - listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "", "") + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "", "", 0) defer listener.Close() reqPacket1 := firstPacket([]byte("POST / HTTP/1.1\r\nTransfer-Encoding: chunked\r\nExpect: 100-continue\r\n\r\n")) @@ -532,7 +532,7 @@ func getMessage() []*TCPPacket { // Response comes before Request func TestRawListenerBench(t *testing.T) { - l := NewListener("", "0", EnginePcap, true, 200*time.Millisecond, "", "") + l := NewListener("", "0", EnginePcap, true, 200*time.Millisecond, "", "", 0) defer l.Close() // Should re-construct message from all possible combinations @@ -583,7 +583,7 @@ func TestRawListenerBench(t *testing.T) { func TestResponseZeroContentLength(t *testing.T) { var req, resp *TCPMessage - listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "", "") + listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond, "", "", 0) defer listener.Close() reqPacket := firstPacket([]byte("POST /api/setup/install HTTP/1.1\r\nHost: localhost:22936\r\nUser-Agent: curl/7.57.0\r\nAccept: */*\r\nContent-Length: 0\r\nContent-Type: application/x-www-form-urlencoded\r\n\r\n")) diff --git a/raw_socket_listener/tcp_message.go b/raw_socket_listener/tcp_message.go index 964770c..cb15b9b 100644 --- a/raw_socket_listener/tcp_message.go +++ b/raw_socket_listener/tcp_message.go @@ -370,10 +370,10 @@ func (t *TCPMessage) updateBodyType() { return case httpMethodKnown: - if ! t.IsIncoming && + if !t.IsIncoming && t.AssocMessage != nil && - bytes.IndexByte( t.AssocMessage.Bytes(), ' ') > -1 && - bytes.Equal( []byte("HEAD"), proto.Method(t.AssocMessage.Bytes()) ) { + bytes.IndexByte(t.AssocMessage.Bytes(), ' ') > -1 && + bytes.Equal([]byte("HEAD"), proto.Method(t.AssocMessage.Bytes())) { // Need to check if this is a response to a head request, // in which case the body has to be empty regardless. t.bodyType = httpBodyEmpty diff --git a/settings.go b/settings.go index 3ae0204..edf7ba6 100644 --- a/settings.go +++ b/settings.go @@ -55,6 +55,7 @@ type AppSettings struct { inputRAWExpire time.Duration inputRAWBpfFilter string inputRAWTimestampType string + inputRawBufferSize int middleware string @@ -133,6 +134,8 @@ func init() { flag.StringVar(&Settings.inputRAWTimestampType, "input-raw-timestamp-type", "", "Possible values: PCAP_TSTAMP_HOST, PCAP_TSTAMP_HOST_LOWPREC, PCAP_TSTAMP_HOST_HIPREC, PCAP_TSTAMP_ADAPTER, PCAP_TSTAMP_ADAPTER_UNSYNCED. This values not supported on all systems, GoReplay will tell you available values of you put wrong one.") + flag.IntVar(&Settings.inputRawBufferSize, "input-raw-buffer-size", 0, "Controls size of the OS buffer (in bytes) which holds packets until they dispatched. Default value depends by system: in Linux around 2MB. If you see big package drop, increase this value.") + flag.StringVar(&Settings.middleware, "middleware", "", "Used for modifying traffic using external command") // flag.Var(&Settings.inputHTTP, "input-http", "Read requests from HTTP, should be explicitly sent from your application:\n\t# Listen for http on 9000\n\tgor --input-http :9000 --output-http staging.com") diff --git a/vendor/vendor.json b/vendor/vendor.json index a6319c5..5db34eb 100644 --- a/vendor/vendor.json +++ b/vendor/vendor.json @@ -45,10 +45,10 @@ "revisionTime": "2016-05-29T05:00:41Z" }, { - "checksumSHA1": "U2Ydh7vEAKlN0Wq22n1JpefF7uY=", + "checksumSHA1": "WT6lYgJhoWbXLpnFOxPISxrL2/o=", "path": "github.com/google/gopacket", - "revision": "b09bf408520f7646e29b7033d9adb00ed779a1c4", - "revisionTime": "2016-05-12T15:06:07Z" + "revision": "60ab61cd59496fcfa4d208b265ba79b1e37c1476", + "revisionTime": "2018-05-13T17:29:36Z" }, { "checksumSHA1": "BM6ZlNJmtKy3GBoWwg2X55gnZ4A=",