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

Next batch

Browse Source
This commit is contained in:
Leonid Bugaev
2015-07-17 11:30:43 +05:00
parent 28540f43a1
commit 125d9db834
17 changed files with 82 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.travis.yml
+2 -2
View File
@@ -1,3 +1,3 @@
language: go
language: go
go: 1.4.2
script: sudo -E bash -c "source /etc/profile && eval '$(gimme 1.4)' && export GOPATH=$HOME/gopath:$GOPATH && go get && GORACE='halt_on_error=1' go test -race -v"
script: sudo -E bash -c "source /etc/profile && eval '$(gimme 1.4.2)' && export GOPATH=$HOME/gopath:$GOPATH && go get && GORACE='halt_on_error=1' go test -v"
emitter.go
+2 -1
View File
@@ -5,6 +5,7 @@ import (
"time"
)
// Start initialize loop for sending data from inputs to outputs
func Start(stop chan int) {
for _, in := range Plugins.Inputs {
go CopyMulty(in, Plugins.Outputs...)
@@ -19,7 +20,7 @@ func Start(stop chan int) {
}
}
// Copy from 1 reader to multiple writers
// CopyMulty copies from 1 reader to multiple writers
func CopyMulty(src io.Reader, writers ...io.Writer) (err error) {
buf := make([]byte, 5*1024*1024)
wIndex := 0
http_client.go
+1 -1
View File
@@ -145,7 +145,7 @@ func (c *HTTPClient) Send(data []byte) (response []byte, err error) {
// 3xx requests
if status[0] == '3' {
c.redirectsCount += 1
c.redirectsCount++
location := proto.Header(payload, []byte("Location"))
redirectPayload := []byte("GET " + string(location) + " HTTP/1.1\r\n\r\n")
http_client_test.go
+12 -12
View File
@@ -132,7 +132,7 @@ func TestHTTPClientHTTPSSend(t *testing.T) {
func TestHTTPClientServerInstantDisconnect(t *testing.T) {
wg := new(sync.WaitGroup)
GET_payload := []byte("GET / HTTP/1.1\r\n\r\n")
GETPayload := []byte("GET / HTTP/1.1\r\n\r\n")
ln, _ := net.Listen("tcp", ":0")
@@ -148,8 +148,8 @@ func TestHTTPClientServerInstantDisconnect(t *testing.T) {
client := NewHTTPClient(ln.Addr().String(), &HTTPClientConfig{})
wg.Add(2)
client.Send(GET_payload)
client.Send(GET_payload)
client.Send(GETPayload)
client.Send(GETPayload)
wg.Wait()
}
@@ -157,7 +157,7 @@ func TestHTTPClientServerInstantDisconnect(t *testing.T) {
func TestHTTPClientServerNoKeepAlive(t *testing.T) {
wg := new(sync.WaitGroup)
GET_payload := []byte("GET / HTTP/1.1\r\n\r\n")
GETPayload := []byte("GET / HTTP/1.1\r\n\r\n")
ln, _ := net.Listen("tcp", ":0")
@@ -186,8 +186,8 @@ func TestHTTPClientServerNoKeepAlive(t *testing.T) {
client := NewHTTPClient(ln.Addr().String(), &HTTPClientConfig{})
wg.Add(2)
client.Send(GET_payload)
client.Send(GET_payload)
client.Send(GETPayload)
client.Send(GETPayload)
wg.Wait()
}
@@ -195,7 +195,7 @@ func TestHTTPClientServerNoKeepAlive(t *testing.T) {
func TestHTTPClientRedirect(t *testing.T) {
wg := new(sync.WaitGroup)
GET_payload := []byte("GET / HTTP/1.1\r\n\r\n")
GETPayload := []byte("GET / HTTP/1.1\r\n\r\n")
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -210,7 +210,7 @@ func TestHTTPClientRedirect(t *testing.T) {
// Should do 2 queries
wg.Add(2)
client.Send(GET_payload)
client.Send(GETPayload)
wg.Wait()
}
@@ -218,7 +218,7 @@ func TestHTTPClientRedirect(t *testing.T) {
func TestHTTPClientRedirectLimit(t *testing.T) {
wg := new(sync.WaitGroup)
GET_payload := []byte("GET / HTTP/1.1\r\n\r\n")
GETPayload := []byte("GET / HTTP/1.1\r\n\r\n")
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -241,7 +241,7 @@ func TestHTTPClientRedirectLimit(t *testing.T) {
// Have 3 redirects + 1 GET, but should do only 2 redirects + GET
wg.Add(3)
client.Send(GET_payload)
client.Send(GETPayload)
wg.Wait()
}
@@ -249,7 +249,7 @@ func TestHTTPClientRedirectLimit(t *testing.T) {
func TestHTTPClientHandleHTTP10(t *testing.T) {
wg := new(sync.WaitGroup)
GET_payload := []byte("GET http://foobar.com/path HTTP/1.0\r\n\r\n")
GETPayload := []byte("GET http://foobar.com/path HTTP/1.0\r\n\r\n")
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -263,7 +263,7 @@ func TestHTTPClientHandleHTTP10(t *testing.T) {
client := NewHTTPClient(server.URL, &HTTPClientConfig{Debug: true})
wg.Add(1)
client.Send(GET_payload)
client.Send(GETPayload)
wg.Wait()
}
http_modifier_settings.go
+4 -2
View File
@@ -8,6 +8,7 @@ import (
"strings"
)
// HTTPModifierConfig holds configuration options for built-in traffic modifier
type HTTPModifierConfig struct {
urlNegativeRegexp HTTPUrlRegexp
urlRegexp HTTPUrlRegexp
@@ -29,6 +30,7 @@ type headerFilter struct {
regexp *regexp.Regexp
}
// HTTPHeaderFilters holds list of headers and their regexps
type HTTPHeaderFilters []headerFilter
func (h *HTTPHeaderFilters) String() string {
@@ -67,7 +69,7 @@ func (h *HTTPHashFilters) String() string {
func (h *HTTPHashFilters) Set(value string) error {
valArr := strings.SplitN(value, ":", 2)
if len(valArr) < 2 {
return errors.New("need both header and value, colon-delimited (ex. user_id:50%).")
return errors.New("need both header and value, colon-delimited (ex. user_id:50%)")
}
f := hashFilter{name: []byte(valArr[0])}
@@ -180,7 +182,7 @@ func (r *UrlRewriteMap) String() string {
func (r *UrlRewriteMap) Set(value string) error {
valArr := strings.SplitN(value, ":", 2)
if len(valArr) < 2 {
return errors.New("need both src and target, colon-delimited (ex. /a:/b).")
return errors.New("need both src and target, colon-delimited (ex. /a:/b)")
}
regexp, err := regexp.Compile(valArr[0])
if err != nil {
http_modifier_test.go
+11 -11
View File
@@ -40,7 +40,7 @@ func TestHTTPModifierHeaderFilters(t *testing.T) {
}
func TestHTTPModifierURLRewrite(t *testing.T) {
var url, new_url []byte
var url, newURL []byte
rewrites := UrlRewriteMap{}
@@ -58,13 +58,13 @@ func TestHTTPModifierURLRewrite(t *testing.T) {
})
url = []byte("/v1/user/joe/ping")
if new_url = proto.Path(modifier.Rewrite(payload(url))); bytes.Equal(new_url, url) {
t.Error("Request url should have been rewritten, wasn't", string(new_url))
if newURL = proto.Path(modifier.Rewrite(payload(url))); bytes.Equal(newURL, url) {
t.Error("Request url should have been rewritten, wasn't", string(newURL))
}
url = []byte("/v1/user/ping")
if new_url = proto.Path(modifier.Rewrite(payload(url))); !bytes.Equal(new_url, url) {
t.Error("Request url should have been rewritten, wasn't", string(new_url))
if newURL = proto.Path(modifier.Rewrite(payload(url))); !bytes.Equal(newURL, url) {
t.Error("Request url should have been rewritten, wasn't", string(newURL))
}
}
@@ -128,9 +128,9 @@ func TestHTTPModifierHeaders(t *testing.T) {
})
payload := []byte("POST /post HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2")
new_payload := []byte("POST /post HTTP/1.1\r\nHeader1: 1\r\nContent-Length: 7\r\nHost: localhost\r\n\r\na=1&b=2")
newPayload := []byte("POST /post HTTP/1.1\r\nHeader1: 1\r\nContent-Length: 7\r\nHost: localhost\r\n\r\na=1&b=2")
if payload = modifier.Rewrite(payload); !bytes.Equal(payload, new_payload) {
if payload = modifier.Rewrite(payload); !bytes.Equal(payload, newPayload) {
t.Error("Should update request headers", string(payload))
}
}
@@ -196,9 +196,9 @@ func TestHTTPModifierSetHeader(t *testing.T) {
})
payload := []byte("POST /post HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2")
payload_after := []byte("POST /post HTTP/1.1\r\nUser-Agent: Gor\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2")
payloadAfter := []byte("POST /post HTTP/1.1\r\nUser-Agent: Gor\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2")
if payload = modifier.Rewrite(payload); !bytes.Equal(payload_after, payload) {
if payload = modifier.Rewrite(payload); !bytes.Equal(payloadAfter, payload) {
t.Error("Should add new header", string(payload))
}
}
@@ -212,9 +212,9 @@ func TestHTTPModifierSetParam(t *testing.T) {
})
payload := []byte("POST /post?api_key=1234 HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2")
payload_after := []byte("POST /post?api_key=1 HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2")
payloadAfter := []byte("POST /post?api_key=1 HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.org\r\n\r\na=1&b=2")
if payload = modifier.Rewrite(payload); !bytes.Equal(payload_after, payload) {
if payload = modifier.Rewrite(payload); !bytes.Equal(payloadAfter, payload) {
t.Error("Should override param", string(payload))
}
}
input_dummy.go
+2
View File
@@ -4,10 +4,12 @@ import (
"time"
)
// DummyInput used for debugging. It generate 1 "GET /"" request per second.
type DummyInput struct {
data chan []byte
}
// NewDummyInput constructor for DummyInput
func NewDummyInput(options string) (di *DummyInput) {
di = new(DummyInput)
di.data = make(chan []byte)
input_file.go
+4 -2
View File
@@ -7,6 +7,7 @@ import (
"time"
)
// FileInput can read requests generated by FileOutput
type FileInput struct {
data chan []byte
path string
@@ -14,19 +15,20 @@ type FileInput struct {
speedFactor float64
}
// NewFileInput constructor for FileInput. Accepts file path as argument.
func NewFileInput(path string) (i *FileInput) {
i = new(FileInput)
i.data = make(chan []byte)
i.path = path
i.speedFactor = 1
i.Init(path)
i.init(path)
go i.emit()
return
}
func (i *FileInput) Init(path string) {
func (i *FileInput) init(path string) {
file, err := os.Open(path)
if err != nil {
input_http.go
+2
View File
@@ -7,12 +7,14 @@ import (
"net/http/httputil"
)
// HTTPInput used for sending requests to Gor via http
type HTTPInput struct {
data chan []byte
address string
listener net.Listener
}
// NewHTTPInput constructor for HTTPInput. Accepts address with port which he will listen on.
func NewHTTPInput(address string) (i *HTTPInput) {
i = new(HTTPInput)
i.data = make(chan []byte)
input_raw.go
+2
View File
@@ -7,11 +7,13 @@ import (
"strings"
)
// RAWInput used for intercepting traffic for given address
type RAWInput struct {
data chan []byte
address string
}
// NewRAWInput constructor for RAWInput. Accepts address with port as argument.
func NewRAWInput(address string) (i *RAWInput) {
i = new(RAWInput)
i.data = make(chan []byte)
input_raw_test.go
+22 -22
View File
@@ -49,7 +49,7 @@ func TestInputRAW100Expect(t *testing.T) {
wg := new(sync.WaitGroup)
quit := make(chan int)
file_content, _ := ioutil.ReadFile("README.md")
fileContent, _ := ioutil.ReadFile("README.md")
// Origing and Replay server initialization
origin := startHTTP(func(req *http.Request) {
@@ -59,12 +59,12 @@ func TestInputRAW100Expect(t *testing.T) {
wg.Done()
})
origin_address := strings.Replace(origin.Addr().String(), "[::]", "127.0.0.1", -1)
originAddr := strings.Replace(origin.Addr().String(), "[::]", "127.0.0.1", -1)
input := NewRAWInput(origin_address)
input := NewRAWInput(originAddr)
// We will use it to get content of raw HTTP request
test_output := NewTestOutput(func(data []byte) {
testOutput := NewTestOutput(func(data []byte) {
if strings.Contains(string(data), "Expect: 100-continue") {
t.Error("Should not contain 100-continue header")
}
@@ -75,24 +75,24 @@ func TestInputRAW100Expect(t *testing.T) {
defer req.Body.Close()
body, _ := ioutil.ReadAll(req.Body)
if !bytes.Equal(body, file_content) {
if !bytes.Equal(body, fileContent) {
buf, _ := httputil.DumpRequest(req, true)
t.Error("Wrong POST body:", string(buf))
}
wg.Done()
})
replay_address := listener.Addr().String()
replayAddr := listener.Addr().String()
http_output := NewHTTPOutput(replay_address, &HTTPOutputConfig{})
httpOutput := NewHTTPOutput(replayAddr, &HTTPOutputConfig{})
Plugins.Inputs = []io.Reader{input}
Plugins.Outputs = []io.Writer{test_output, http_output}
Plugins.Outputs = []io.Writer{testOutput, httpOutput}
go Start(quit)
wg.Add(3)
curl := exec.Command("curl", "http://"+origin_address, "--data-binary", "@README.md")
curl := exec.Command("curl", "http://"+originAddr, "--data-binary", "@README.md")
err := curl.Run()
if err != nil {
log.Fatal(err)
@@ -106,7 +106,7 @@ func TestInputRAWChunkedEncoding(t *testing.T) {
wg := new(sync.WaitGroup)
quit := make(chan int)
file_content, _ := ioutil.ReadFile("README.md")
fileContent, _ := ioutil.ReadFile("README.md")
// Origing and Replay server initialization
origin := startHTTP(func(req *http.Request) {
@@ -116,33 +116,33 @@ func TestInputRAWChunkedEncoding(t *testing.T) {
wg.Done()
})
origin_address := strings.Replace(origin.Addr().String(), "[::]", "127.0.0.1", -1)
originAddr := strings.Replace(origin.Addr().String(), "[::]", "127.0.0.1", -1)
input := NewRAWInput(origin_address)
input := NewRAWInput(originAddr)
listener := startHTTP(func(req *http.Request) {
defer req.Body.Close()
body, _ := ioutil.ReadAll(req.Body)
if !bytes.Equal(body, file_content) {
if !bytes.Equal(body, fileContent) {
buf, _ := httputil.DumpRequest(req, true)
t.Error("Wrong POST body:", string(buf))
}
wg.Done()
})
replay_address := listener.Addr().String()
replayAddr := listener.Addr().String()
http_output := NewHTTPOutput(replay_address, &HTTPOutputConfig{Debug: true})
httpOutput := NewHTTPOutput(replayAddr, &HTTPOutputConfig{Debug: true})
Plugins.Inputs = []io.Reader{input}
Plugins.Outputs = []io.Writer{http_output}
Plugins.Outputs = []io.Writer{httpOutput}
go Start(quit)
wg.Add(2)
curl := exec.Command("curl", "http://"+origin_address, "--header", "Transfer-Encoding: chunked", "--data-binary", "@README.md")
curl := exec.Command("curl", "http://"+originAddr, "--header", "Transfer-Encoding: chunked", "--data-binary", "@README.md")
err := curl.Run()
if err != nil {
log.Fatal(err)
@@ -175,9 +175,9 @@ func TestInputRAWLargePayload(t *testing.T) {
wg.Done()
})
origin_address := strings.Replace(origin.Addr().String(), "[::]", "127.0.0.1", -1)
originAddr := strings.Replace(origin.Addr().String(), "[::]", "127.0.0.1", -1)
input := NewRAWInput(origin_address)
input := NewRAWInput(originAddr)
replay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
req.Body = http.MaxBytesReader(w, req.Body, 1*1024*1024)
@@ -193,15 +193,15 @@ func TestInputRAWLargePayload(t *testing.T) {
}))
defer replay.Close()
http_output := NewHTTPOutput(replay.URL, &HTTPOutputConfig{Debug: false})
httpOutput := NewHTTPOutput(replay.URL, &HTTPOutputConfig{Debug: false})
Plugins.Inputs = []io.Reader{input}
Plugins.Outputs = []io.Writer{http_output}
Plugins.Outputs = []io.Writer{httpOutput}
go Start(quit)
wg.Add(2)
curl := exec.Command("curl", "http://"+origin_address, "--header", "Transfer-Encoding: chunked", "--data-binary", "@/tmp/large")
curl := exec.Command("curl", "http://"+originAddr, "--header", "Transfer-Encoding: chunked", "--data-binary", "@/tmp/large")
err = curl.Run()
if err != nil {
log.Fatal("curl error:", err)
input_tcp.go
+3 -3
View File
@@ -9,15 +9,15 @@ import (
"os"
)
// Can be tested using nc tool:
// echo "asdad" | nc 127.0.0.1 27017
//
// TCPInput used for internal communication
// It expected hex encoded data
type TCPInput struct {
data chan []byte
address string
listener net.Listener
}
// NewTCPInput constructor for TCPInput, accepts address with port
func NewTCPInput(address string) (i *TCPInput) {
i = new(TCPInput)
i.data = make(chan []byte)
input_tcp_test.go
+3 -3
View File
@@ -83,10 +83,10 @@ func BenchmarkTCPInput(b *testing.B) {
for {
data := <-dataChan
new_buf := make([]byte, len(data)+2)
buf := make([]byte, len(data)+2)
data = append(data, []byte("¶")...)
copy(new_buf, data)
conn.Write(new_buf)
copy(buf, data)
conn.Write(buf)
}
}(conn)
}
limiter.go
+3
View File
@@ -9,6 +9,7 @@ import (
"time"
)
// Limiter is a wrapper for input or output plugin which adds rate limiting
type Limiter struct {
plugin interface{}
limit int
@@ -30,6 +31,8 @@ func parseLimitOptions(options string) (limit int, isPercent bool) {
return
}
// NewLimiter constructor for Limiter, accepts plugin and options
// `options` allow to sprcify relatve or absolute limiting
func NewLimiter(plugin interface{}, options string) io.ReadWriter {
l := new(Limiter)
l.limit, l.isPercent = parseLimitOptions(options)
output_dummy.go
+2
View File
@@ -4,9 +4,11 @@ import (
"fmt"
)
// DummyOutput used for debugging, prints all incoming requests
type DummyOutput struct {
}
// NewDummyOutput constructor for DummyOutput
func NewDummyOutput(options string) (di *DummyOutput) {
di = new(DummyOutput)
output_file.go
+6 -2
View File
@@ -8,26 +8,30 @@ import (
"time"
)
// RawRequest stores original start time and request payload
type RawRequest struct {
Timestamp int64
Request []byte
}
// FileOutput output plugin
type FileOutput struct {
path string
encoder *gob.Encoder
file *os.File
}
// NewFileOutput constructor for FileOutput, accepts path
func NewFileOutput(path string) io.Writer {
o := new(FileOutput)
o.path = path
o.Init(path)
o.init(path)
return o
}
func (o *FileOutput) Init(path string) {
func (o *FileOutput) init(path string) {
var err error
o.file, err = os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0660)
plugins.go
+1 -1
View File
@@ -13,7 +13,7 @@ type InOutPlugins struct {
}
// Plugins holds all the plugin objects
var Plugins *InOutPlugins
var Plugins *InOutPlugins = new(InOutPlugins)
// extractLimitOptions detects if plugin get called with limiter support
// Returns address and limit