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

Merge branch 'master' of https://github.com/buger/gor

Browse Source
This commit is contained in:
Leonid Bugaev
2017-01-23 20:04:33 +03:00
parent 973b028517 d66ba0353a
commit 1a32de3dc6
15 changed files with 170 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
emitter.go
+14 -4
View File
@@ -28,6 +28,12 @@ func Start(stop chan int) {
for _, in := range Plugins.Inputs {
go CopyMulty(in, Plugins.Outputs...)
}
for _, out := range Plugins.Outputs {
if r, ok := out.(io.Reader); ok {
go CopyMulty(r, Plugins.Outputs...)
}
}
}
for {
@@ -89,12 +95,16 @@ func CopyMulty(src io.Reader, writers ...io.Writer) (err error) {
}
} else {
if _, ok := filteredRequests[requestID]; ok {
delete(filteredRequests, requestID);
delete(filteredRequests, requestID)
continue
}
}
}
if Settings.prettifyHTTP {
payload = prettifyHTTP(payload)
}
if Settings.splitOutput {
if Settings.recognizeTCPSessions {
hasher := fnv.New32a()
@@ -130,12 +140,12 @@ func CopyMulty(src io.Reader, writers ...io.Writer) (err error) {
}
// Run GC on each 1000 request
if i % 1000 == 0 {
if i%1000 == 0 {
// Clean up filtered requests for which we didn't get a response to filter
now := time.Now()
if now.Sub(filteredRequestsLastCleanTime) > 60 * time.Second {
if now.Sub(filteredRequestsLastCleanTime) > 60*time.Second {
for k, v := range filteredRequests {
if now.Sub(v) > 60 * time.Second {
if now.Sub(v) > 60*time.Second {
delete(filteredRequests, k)
}
}
emitter_test.go
+2
View File
@@ -76,6 +76,8 @@ func TestEmitterFiltered(t *testing.T) {
wg.Wait()
close(quit)
Settings.modifierConfig = HTTPModifierConfig{}
}
func TestEmitterSplitRoundRobin(t *testing.T) {
http_prettifier.go
+68
View File
@@ -0,0 +1,68 @@
package main
import (
"github.com/buger/gor/proto"
"bytes"
"compress/gzip"
"strconv"
"io/ioutil"
"net/http/httputil"
)
func prettifyHTTP(p []byte) []byte {
headSize := bytes.IndexByte(p, '\n') + 1
head := p[:headSize]
body := p[headSize:]
headersPos := proto.MIMEHeadersEndPos(body)
headers := body[:headersPos]
content := body[headersPos:]
var tEnc, cEnc []byte
proto.ParseHeaders([][]byte{headers}, func(header, value []byte) bool {
if proto.HeadersEqual(header, []byte("Transfer-Encoding")) {
tEnc = value
}
if proto.HeadersEqual(header, []byte("Content-Encoding")) {
cEnc = value
}
return true
})
if len(tEnc) == 0 && len(cEnc) == 0 {
return p
}
if bytes.Equal(tEnc, []byte("chunked")) {
buf := bytes.NewBuffer(content)
r := httputil.NewChunkedReader(buf)
content, _ = ioutil.ReadAll(r)
headers = proto.DeleteHeader(headers, []byte("Transfer-Encoding"))
newLen := strconv.Itoa(len(content))
headers = proto.SetHeader(headers, []byte("Content-Length"), []byte(newLen))
}
if bytes.Equal(cEnc, []byte("gzip")) {
buf := bytes.NewBuffer(content)
g, err := gzip.NewReader(buf)
if err != nil {
Debug("[Prettifier] GZIP encoding error:", err)
}
content, _ = ioutil.ReadAll(g)
headers = proto.DeleteHeader(headers, []byte("Content-Encoding"))
newLen := strconv.Itoa(len(content))
headers = proto.SetHeader(headers, []byte("Content-Length"), []byte(newLen))
}
newPayload := append(append(head, headers...), content...)
return newPayload
}
http_prettifier_test.go
+36
View File
@@ -0,0 +1,36 @@
package main
import (
"compress/gzip"
"testing"
"strconv"
"bytes"
)
func TestHTTPPrettifierGzip(t *testing.T) {
b := bytes.NewBufferString("")
w := gzip.NewWriter(b)
w.Write([]byte("test"))
w.Close()
size := strconv.Itoa(len(b.Bytes()))
payload := []byte("2 1 1\nHTTP/1.1 200 OK\r\nContent-Length: " + size + "\r\nContent-Encoding: gzip\r\n\r\n")
payload = append(payload, b.Bytes()...)
newPayload := prettifyHTTP(payload)
if string(newPayload) != "2 1 1\nHTTP/1.1 200 OK\r\nContent-Length: 4\r\n\r\ntest" {
t.Error("Payload not match:", string(newPayload))
}
}
func TestHTTPPrettifierChunked(t *testing.T) {
payload := []byte("POST / HTTP/1.1\r\nHost: www.w3.org\r\nTransfer-Encoding: chunked\r\n\r\n4\r\nWiki\r\n5\r\npedia\r\ne\r\n in\r\n\r\nchunks.\r\n0\r\n\r\n")
newPayload := prettifyHTTP(payload)
if string(newPayload) != "POST / HTTP/1.1\r\nHost: www.w3.org\r\nContent-Length: 23\r\n\r\nWikipedia in\r\n\r\nchunks." {
t.Error("Payload not match:", string(newPayload))
}
}
input_kafka_test.go
+2 -2
View File
@@ -12,7 +12,7 @@ func TestInputKafkaRAW(t *testing.T) {
consumer.ExpectConsumePartition("test", 0, mocks.AnyOffset).YieldMessage(&sarama.ConsumerMessage{Value: []byte("1 2 3\nGET / HTTP1.1\r\nHeader: 1\r\n\r\n")})
consumer.SetTopicMetadata(
map[string][]int32{"test": []int32{0}},
map[string][]int32{"test": {0}},
)
input := NewKafkaInput("", &KafkaConfig{
@@ -39,7 +39,7 @@ func TestInputKafkaJSON(t *testing.T) {
consumer.ExpectConsumePartition("test", 0, mocks.AnyOffset).YieldMessage(&sarama.ConsumerMessage{Value: []byte(`{"Req_URL":"/","Req_Type":"1","Req_ID":"2","Req_Ts":"3","Req_Method":"GET","Req_Headers":{"Header":"1"}}`)})
consumer.SetTopicMetadata(
map[string][]int32{"test": []int32{0}},
map[string][]int32{"test": {0}},
)
input := NewKafkaInput("", &KafkaConfig{
limiter.go
+5 -1
View File
@@ -82,7 +82,11 @@ func (l *Limiter) Write(data []byte) (n int, err error) {
}
func (l *Limiter) Read(data []byte) (n int, err error) {
n, err = l.plugin.(io.Reader).Read(data)
if r, ok := l.plugin.(io.Reader); ok {
n, err = r.Read(data)
} else {
return 0, nil
}
if l.isLimited() {
return 0, nil
output_file.go
+15 -7
View File
@@ -24,7 +24,8 @@ var dateFileNameFuncs = map[string]func(*FileOutput) string{
"%M": func(o *FileOutput) string { return time.Now().Format("04") },
"%S": func(o *FileOutput) string { return time.Now().Format("05") },
"%NS": func(o *FileOutput) string { return fmt.Sprint(time.Now().Nanosecond()) },
"%r": func(o *FileOutput) string { return o.currentID },
"%r": func(o *FileOutput) string { return string(o.currentID) },
"%t": func(o *FileOutput) string { return string(o.payloadType) },
}
type FileOutputConfig struct {
@@ -45,7 +46,9 @@ type FileOutput struct {
chunkSize int
writer io.Writer
requestPerFile bool
currentID string
currentID []byte
payloadType []byte
closed bool
config *FileOutputConfig
}
@@ -68,8 +71,13 @@ func NewFileOutput(pathTemplate string, config *FileOutputConfig) *FileOutput {
go func() {
for {
time.Sleep(config.flushInterval)
if o.closed {
break
}
o.mu.Lock()
o.updateName()
o.flush()
o.mu.Unlock()
}
}()
@@ -181,14 +189,12 @@ func (o *FileOutput) updateName() {
func (o *FileOutput) Write(data []byte) (n int, err error) {
if o.requestPerFile {
o.currentID = string(payloadMeta(data)[1])
meta := payloadMeta(data)
o.currentID = meta[1]
o.payloadType = meta[0]
o.updateName()
}
if !isOriginPayload(data) {
return len(data), nil
}
if o.file == nil || o.currentName != o.file.Name() {
o.mu.Lock()
o.Close()
@@ -259,5 +265,7 @@ func (o *FileOutput) Close() error {
go o.config.onClose(o.file.Name())
}
}
o.closed = true
return nil
}
output_file_test.go
+3 -2
View File
@@ -65,9 +65,10 @@ func TestFileOutputWithNameCleaning(t *testing.T) {
}
func TestFileOutputPathTemplate(t *testing.T) {
output := &FileOutput{pathTemplate: "/tmp/log-%Y-%m-%d-%S", config: &FileOutputConfig{flushInterval: time.Minute, append: true}}
output := &FileOutput{pathTemplate: "/tmp/log-%Y-%m-%d-%S-%t", config: &FileOutputConfig{flushInterval: time.Minute, append: true}}
now := time.Now()
expectedPath := fmt.Sprintf("/tmp/log-%s-%s-%s-%s", now.Format("2006"), now.Format("01"), now.Format("02"), now.Format("05"))
output.payloadType = []byte("3")
expectedPath := fmt.Sprintf("/tmp/log-%s-%s-%s-%s-3", now.Format("2006"), now.Format("01"), now.Format("02"), now.Format("05"))
path := output.filename()
if expectedPath != path {
output_http.go
-4
View File
@@ -131,10 +131,6 @@ func NewHTTPOutput(address string, config *HTTPOutputConfig) io.Writer {
o.elasticSearch.Init(o.config.elasticSearch)
}
if len(Settings.middleware) > 0 {
o.config.TrackResponses = true
}
if Settings.recognizeTCPSessions {
o.workerSessions = make(map[string]*httpWorker, 100)
go o.sessionWorkerMaster()
output_http_test.go
+8 -4
View File
@@ -43,15 +43,19 @@ func TestHTTPOutput(t *testing.T) {
methods := HTTPMethods{[]byte("GET"), []byte("PUT"), []byte("POST")}
Settings.modifierConfig = HTTPModifierConfig{headers: headers, methods: methods}
output := NewHTTPOutput(server.URL, &HTTPOutputConfig{Debug: true})
http_output := NewHTTPOutput(server.URL, &HTTPOutputConfig{Debug: true, TrackResponses: true})
output := NewTestOutput(func(data []byte) {
wg.Done()
})
Plugins.Inputs = []io.Reader{input}
Plugins.Outputs = []io.Writer{output}
Plugins.Outputs = []io.Writer{http_output, output}
go Start(quit)
for i := 0; i < 100; i++ {
wg.Add(2) // OPTIONS should be ignored
for i := 0; i < 1; i++ {
// 2 http-output, 2 - test output request, 2 - test output http response
wg.Add(6) // OPTIONS should be ignored
input.EmitPOST()
input.EmitOPTIONS()
input.EmitGET()
proto/fuzz.go
+4 -4
View File
@@ -4,9 +4,9 @@ package proto
func Fuzz(data []byte) int {
ParseHeaders([][]byte{data}, func(header []byte, value []byte) bool {
return true
})
ParseHeaders([][]byte{data}, func(header []byte, value []byte) bool {
return true
})
return 1
return 1
}
proto/proto.go
+3 -3
View File
@@ -32,7 +32,7 @@ var HeaderDelim = []byte(": ")
// MIMEHeadersEndPos finds end of the Headers section, which should end with empty line.
func MIMEHeadersEndPos(payload []byte) int {
return bytes.Index(payload, EmptyLine)
return bytes.Index(payload, EmptyLine) + 4
}
// MIMEHeadersStartPos finds start of Headers section
@@ -181,7 +181,7 @@ func HeadersEqual(h1 []byte, h2 []byte) bool {
// Parsing headers from multiple payloads
func ParseHeaders(payloads [][]byte, cb func(header []byte, value []byte) bool) {
hS := [2]int{0, 0} // header start
hS := [2]int{0, 0} // header start
hE := [2]int{-1, -1} // header end
vS := [2]int{-1, -1} // value start
vE := [2]int{-1, -1} // value end
@@ -337,7 +337,7 @@ func DeleteHeader(payload, name []byte) []byte {
// Body returns request/response body
func Body(payload []byte) []byte {
// 4 -> len(EMPTY_LINE)
return payload[MIMEHeadersEndPos(payload)+4:]
return payload[MIMEHeadersEndPos(payload):]
}
// Path takes payload and retuns request path: Split(firstLine, ' ')[1]
proto/proto_test.go
+2 -2
View File
@@ -153,8 +153,8 @@ func TestFuzzCrashers(t *testing.T) {
for _, f := range crashers {
ParseHeaders([][]byte{[]byte(f)}, func(header []byte, value []byte) bool {
return true
})
return true
})
}
}
settings.go
+7 -1
View File
@@ -59,8 +59,9 @@ type AppSettings struct {
middleware string
inputHTTP MultiOption
outputHTTP MultiOption
prettifyHTTP bool
outputHTTPConfig HTTPOutputConfig
outputBinary MultiOption
@@ -116,10 +117,14 @@ func init() {
flag.Var(&Settings.outputFileConfig.sizeLimit, "output-file-size-limit", "Size of each chunk. Default: 32mb")
flag.IntVar(&Settings.outputFileConfig.queueLimit, "output-file-queue-limit", 256, "The length of the chunk queue. Default: 256")
<<<<<<< HEAD
flag.Var(&Settings.outputS3, "output-s3", "Write incoming requests to S3 path: \n\tgor --input-raw :80 --output-s3 s3://mybucket/logs/%Y-%m-%d.gz")
flag.StringVar(&Settings.outputS3Config.bufferPath, "output-s3-buffer-path", "/tmp", "The path prefix of the S3 log buffer files.: \n\tgor --input-raw :80 --output-s3 s3://mybucket/logs/%Y-%m-%d.gz --output-s3-buffer-path /mnt/logs")
flag.StringVar(&Settings.outputS3Config.region, "output-s3-region", "us-east-1", "Specify S3 region, default is 'us-east-1': \n\tgor --input-raw :80 --output-s3 s3://mybucket/logs/%Y-%m-%d.gz --output-s3-region us-west-2")
flag.StringVar(&Settings.outputS3Config.endpoint, "output-s3-endpoint", "", "You can specify custom endpoint if using alternative S3 compatitable storage.")
=======
flag.BoolVar(&Settings.prettifyHTTP, "prettify-http", false, "If enabled, will automatically decode requests and responses with: Content-Encodning: gzip and Transfer-Encoding: chunked. Useful for debugging, in conjuction with --output-stdout")
>>>>>>> d66ba0353a9995e2d3be3bca00285b6cb8c859c4
flag.Var(&Settings.inputRAW, "input-raw", "Capture traffic from given port (use RAW sockets and require *sudo* access):\n\t# Capture traffic from 8080 port\n\tgor --input-raw :8080 --output-http staging.com")
@@ -142,6 +147,7 @@ func init() {
flag.IntVar(&Settings.outputHTTPConfig.workers, "output-http-workers", 0, "Gor uses dynamic worker scaling by default. Enter a number to run a set number of workers.")
flag.IntVar(&Settings.outputHTTPConfig.redirectLimit, "output-http-redirects", 0, "Enable how often redirects should be followed.")
flag.DurationVar(&Settings.outputHTTPConfig.Timeout, "output-http-timeout", 5*time.Second, "Specify HTTP request/response timeout. By default 5s. Example: --output-http-timeout 30s")
flag.BoolVar(&Settings.outputHTTPConfig.TrackResponses, "output-http-track-response", false, "If turned on, HTTP output responses will be set to all outputs like stdout, file and etc.")
flag.BoolVar(&Settings.outputHTTPConfig.stats, "output-http-stats", false, "Report http output queue stats to console every 5 seconds.")
flag.BoolVar(&Settings.outputHTTPConfig.OriginalHost, "http-original-host", false, "Normally gor replaces the Host http header with the host supplied with --output-http. This option disables that behavior, preserving the original Host header.")
test_input.go
+1 -1
View File
@@ -8,7 +8,7 @@ import (
// TestInput used for testing purpose, it allows emitting requests on demand
type TestInput struct {
data chan []byte
data chan []byte
skipHeader bool
}