github/goreplay
4
0
Fork 0
mirror of https://github.com/buger/goreplay.git synced 2024-04-21 12:32:02 +00:00
Code Issues Releases Wiki Activity

Add way to control packet capture buffer size and optimize snaplen

Browse Source 
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.
This commit is contained in:
Leonid Bugaev
2018-05-27 12:31:47 +03:00
parent 18610d3555
commit c5d1112e7a
13 changed files with 59 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
elasticsearch_test.go
+1 -1
View File
@@ -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)
}
}
http_client.go
+2 -2
View File
@@ -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
input_kafka.go
+1 -1
View File
@@ -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)
input_raw.go
+4 -2
View File
@@ -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()
input_raw_test.go
+7 -7
View File
@@ -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) {
limiter.go
+1 -1
View File
@@ -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)
}
middleware_test.go
+2 -2
View File
@@ -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
plugins.go
+1 -1
View File
@@ -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 {
raw_socket_listener/listener.go
+16 -2
View File
@@ -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)
raw_socket_listener/listener_test.go
+15 -15
View File
@@ -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"))
raw_socket_listener/tcp_message.go
+3 -3
View File
@@ -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
settings.go
+3
View File
@@ -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")
vendor/vendor.json
Vendored
+3 -3
View File
@@ -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=",