From 40a48863fbffe3ce06ba126480d1becae1a133c1 Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Mon, 5 Aug 2013 18:05:40 +0400 Subject: [PATCH 1/7] Refactored tests --- integration_test.go | 51 +++++++++++++++++++++++++++------------------ 1 file changed, 31 insertions(+), 20 deletions(-) diff --git a/integration_test.go b/integration_test.go index 5f97347..6ec349d 100644 --- a/integration_test.go +++ b/integration_test.go @@ -8,16 +8,23 @@ import ( "time" + "fmt" "net/http" "strconv" ) +func isEqual(t *testing.T, a interface{}, b interface{}) { + if a != b { + t.Error("Original and Replayed request not match\n", a, "!=", b) + } +} + func startListener() { listener.Settings.Verbose = true listener.Settings.Address = "127.0.0.1" listener.Settings.ReplayAddress = "127.0.0.1:50001" listener.Settings.Port = 50000 - go listener.Run() + listener.Run() } func startReplay() { @@ -25,11 +32,11 @@ func startReplay() { replay.Settings.Host = "127.0.0.1" replay.Settings.ForwardAddress = "127.0.0.1:50002" replay.Settings.Port = 50001 - go replay.Run() + replay.Run() } func startHTTP(port int, handler http.Handler) { - go http.ListenAndServe(":"+strconv.Itoa(port), handler) + http.ListenAndServe(":"+strconv.Itoa(port), handler) } func getRequest() *http.Request { @@ -43,35 +50,37 @@ func getRequest() *http.Request { return req } -func TestIntegration(t *testing.T) { +func startEnv(listenHandler http.HandlerFunc, replayHandler http.HandlerFunc) { + go startHTTP(50000, http.HandlerFunc(listenHandler)) + go startListener() + go startReplay() + go startHTTP(50002, http.HandlerFunc(replayHandler)) +} + +func TestReplay(t *testing.T) { request := getRequest() + received := make(chan int) listenHandler := func(w http.ResponseWriter, r *http.Request) { http.Error(w, "404 page not found", http.StatusNotFound) } - startHTTP(50000, http.HandlerFunc(listenHandler)) - - startListener() - startReplay() - - received := make(chan int) replayHandler := func(w http.ResponseWriter, r *http.Request) { - equal := func(a interface{}, b interface{}) { - if a != b { - t.Error("Original and Replayed request not match\n", a, "!=", b, "\nReplayed:", r, "\nOriginal:", request) - } - } - - equal(r.URL.Path, request.URL.Path) - equal(r.Cookies()[0].Value, request.Cookies()[0].Value) + isEqual(t, r.URL.Path, request.URL.Path) + isEqual(t, r.Cookies()[0].Value, request.Cookies()[0].Value) http.Error(w, "404 page not found", http.StatusNotFound) + if t.Failed() { + fmt.Println("\nReplayed:", r, "\nOriginal:", request) + } + received <- 1 } - startHTTP(50002, http.HandlerFunc(replayHandler)) + startEnv(listenHandler, replayHandler) + + // Time to start http and gor instances time.Sleep(time.Millisecond * 100) _, err := http.DefaultClient.Do(request) @@ -85,6 +94,8 @@ func TestIntegration(t *testing.T) { case <-time.After(time.Second): t.Error("Timeout error") } +} - time.Sleep(time.Millisecond * 500) +func TestRateLimit(t *testing.T) { + } From a9ca7ea7ffb460514167a4ba88e52582a5a0b0a0 Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Mon, 5 Aug 2013 18:18:39 +0400 Subject: [PATCH 2/7] Make tests more configurable --- integration_test.go | 33 +++++++++++++++++---------------- 1 file changed, 17 insertions(+), 16 deletions(-) diff --git a/integration_test.go b/integration_test.go index 6ec349d..4229305 100644 --- a/integration_test.go +++ b/integration_test.go @@ -19,19 +19,19 @@ func isEqual(t *testing.T, a interface{}, b interface{}) { } } -func startListener() { - listener.Settings.Verbose = true +func startListener(port int, replayPort int, verbose bool) { + listener.Settings.Verbose = verbose listener.Settings.Address = "127.0.0.1" - listener.Settings.ReplayAddress = "127.0.0.1:50001" - listener.Settings.Port = 50000 + listener.Settings.ReplayAddress = "127.0.0.1:" + strconv.Itoa(replayPort) + listener.Settings.Port = port listener.Run() } -func startReplay() { - replay.Settings.Verbose = true +func startReplay(port int, forwardPort int, verbose bool) { + replay.Settings.Verbose = verbose replay.Settings.Host = "127.0.0.1" - replay.Settings.ForwardAddress = "127.0.0.1:50002" - replay.Settings.Port = 50001 + replay.Settings.ForwardAddress = "127.0.0.1:" + strconv.Itoa(forwardPort) + replay.Settings.Port = port replay.Run() } @@ -50,11 +50,15 @@ func getRequest() *http.Request { return req } -func startEnv(listenHandler http.HandlerFunc, replayHandler http.HandlerFunc) { +func startEnv(listenHandler http.HandlerFunc, replayHandler http.HandlerFunc, verbose bool) { go startHTTP(50000, http.HandlerFunc(listenHandler)) - go startListener() - go startReplay() go startHTTP(50002, http.HandlerFunc(replayHandler)) + + go startListener(50000, 50001, verbose) + go startReplay(50001, 50002, verbose) + + // Time to start http and gor instances + time.Sleep(time.Millisecond * 100) } func TestReplay(t *testing.T) { @@ -62,7 +66,7 @@ func TestReplay(t *testing.T) { received := make(chan int) listenHandler := func(w http.ResponseWriter, r *http.Request) { - http.Error(w, "404 page not found", http.StatusNotFound) + http.Error(w, "OK", http.StatusNotFound) } replayHandler := func(w http.ResponseWriter, r *http.Request) { @@ -78,10 +82,7 @@ func TestReplay(t *testing.T) { received <- 1 } - startEnv(listenHandler, replayHandler) - - // Time to start http and gor instances - time.Sleep(time.Millisecond * 100) + startEnv(listenHandler, replayHandler, true) _, err := http.DefaultClient.Do(request) From e42c3a96e2e52a87cc1fc61d116c61c63630bd5b Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Mon, 5 Aug 2013 18:42:37 +0400 Subject: [PATCH 3/7] Make tests more modular --- integration_test.go | 74 +++++++++++++++++++++++++++++++++------------ 1 file changed, 54 insertions(+), 20 deletions(-) diff --git a/integration_test.go b/integration_test.go index 4229305..99b679a 100644 --- a/integration_test.go +++ b/integration_test.go @@ -19,28 +19,53 @@ func isEqual(t *testing.T, a interface{}, b interface{}) { } } -func startListener(port int, replayPort int, verbose bool) { - listener.Settings.Verbose = verbose +var envs int + +type Env struct { + Verbose bool + + ListenHandler http.HandlerFunc + ReplayHandler http.HandlerFunc +} + +func (e *Env) start() (p int) { + p = 50000 + envs*10 + + go e.startHTTP(p, http.HandlerFunc(e.ListenHandler)) + go e.startHTTP(p+2, http.HandlerFunc(e.ReplayHandler)) + go e.startListener(p, p+1) + go e.startReplay(p+1, p+2) + + // Time to start http and gor instances + time.Sleep(time.Millisecond * 100) + + envs++ + + return +} + +func (e *Env) startListener(port int, replayPort int) { + listener.Settings.Verbose = e.Verbose listener.Settings.Address = "127.0.0.1" listener.Settings.ReplayAddress = "127.0.0.1:" + strconv.Itoa(replayPort) listener.Settings.Port = port listener.Run() } -func startReplay(port int, forwardPort int, verbose bool) { - replay.Settings.Verbose = verbose +func (e *Env) startReplay(port int, forwardPort int) { + replay.Settings.Verbose = e.Verbose replay.Settings.Host = "127.0.0.1" replay.Settings.ForwardAddress = "127.0.0.1:" + strconv.Itoa(forwardPort) replay.Settings.Port = port replay.Run() } -func startHTTP(port int, handler http.Handler) { +func (e *Env) startHTTP(port int, handler http.Handler) { http.ListenAndServe(":"+strconv.Itoa(port), handler) } -func getRequest() *http.Request { - req, _ := http.NewRequest("GET", "http://localhost:50000/test", nil) +func getRequest(port int) *http.Request { + req, _ := http.NewRequest("GET", "http://localhost:"+strconv.Itoa(port)+"/test", nil) ck1 := new(http.Cookie) ck1.Name = "test" ck1.Value = "value" @@ -50,19 +75,8 @@ func getRequest() *http.Request { return req } -func startEnv(listenHandler http.HandlerFunc, replayHandler http.HandlerFunc, verbose bool) { - go startHTTP(50000, http.HandlerFunc(listenHandler)) - go startHTTP(50002, http.HandlerFunc(replayHandler)) - - go startListener(50000, 50001, verbose) - go startReplay(50001, 50002, verbose) - - // Time to start http and gor instances - time.Sleep(time.Millisecond * 100) -} - func TestReplay(t *testing.T) { - request := getRequest() + var request *http.Request received := make(chan int) listenHandler := func(w http.ResponseWriter, r *http.Request) { @@ -82,7 +96,14 @@ func TestReplay(t *testing.T) { received <- 1 } - startEnv(listenHandler, replayHandler, true) + env := &Env{ + Verbose: true, + ListenHandler: listenHandler, + ReplayHandler: replayHandler, + } + p := env.start() + + request = getRequest(p) _, err := http.DefaultClient.Do(request) @@ -98,5 +119,18 @@ func TestReplay(t *testing.T) { } func TestRateLimit(t *testing.T) { + listenHandler := func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "OK", http.StatusAccepted) + } + replayHandler := func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "OK", http.StatusAccepted) + } + + env := &Env{ + ListenHandler: listenHandler, + ReplayHandler: replayHandler, + } + + env.start() } From 00d3103451b19558d2b164d36aa07e39f81e7cdc Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Mon, 5 Aug 2013 19:03:06 +0400 Subject: [PATCH 4/7] Add rate-limit tests --- integration_test.go | 42 ++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 40 insertions(+), 2 deletions(-) diff --git a/integration_test.go b/integration_test.go index 99b679a..612fd12 100644 --- a/integration_test.go +++ b/integration_test.go @@ -11,6 +11,8 @@ import ( "fmt" "net/http" "strconv" + + "sync/atomic" ) func isEqual(t *testing.T, a interface{}, b interface{}) { @@ -26,6 +28,8 @@ type Env struct { ListenHandler http.HandlerFunc ReplayHandler http.HandlerFunc + + ReplayLimit int } func (e *Env) start() (p int) { @@ -57,6 +61,11 @@ func (e *Env) startReplay(port int, forwardPort int) { replay.Settings.Host = "127.0.0.1" replay.Settings.ForwardAddress = "127.0.0.1:" + strconv.Itoa(forwardPort) replay.Settings.Port = port + + if e.ReplayLimit != 0 { + replay.Settings.ForwardAddress += "|" + strconv.Itoa(e.ReplayLimit) + } + replay.Run() } @@ -118,19 +127,48 @@ func TestReplay(t *testing.T) { } } -func TestRateLimit(t *testing.T) { +func rateLimitEnv(limit int) int32 { + var processed int32 + listenHandler := func(w http.ResponseWriter, r *http.Request) { http.Error(w, "OK", http.StatusAccepted) } replayHandler := func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&processed, 1) http.Error(w, "OK", http.StatusAccepted) } env := &Env{ ListenHandler: listenHandler, ReplayHandler: replayHandler, + ReplayLimit: limit, } - env.start() + p := env.start() + req := getRequest(p) + + for i := 0; i < 10; i++ { + http.DefaultClient.Do(req) + } + + time.Sleep(time.Millisecond * 500) + + return processed +} + +func TestWithoutReplayRateLimit(t *testing.T) { + processed := rateLimitEnv(0) + + if processed != 10 { + t.Error("It should forward all requests without rate-limiting", processed) + } +} + +func TestReplayRateLimit(t *testing.T) { + processed := rateLimitEnv(5) + + if processed != 5 { + t.Error("It should forward only 5 requests with rate-limiting", processed) + } } From 77751af316823972726c4749c17b2010e0e89081 Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Mon, 5 Aug 2013 20:42:16 +0400 Subject: [PATCH 5/7] Added listener limit option --- integration_test.go | 25 ++++++++++++++++++++----- listener/listener.go | 23 +++++++++++++++++++++-- listener/settings.go | 14 ++++++++++++++ 3 files changed, 55 insertions(+), 7 deletions(-) diff --git a/integration_test.go b/integration_test.go index 612fd12..c5c44e9 100644 --- a/integration_test.go +++ b/integration_test.go @@ -29,7 +29,8 @@ type Env struct { ListenHandler http.HandlerFunc ReplayHandler http.HandlerFunc - ReplayLimit int + ReplayLimit int + ListenerLimit int } func (e *Env) start() (p int) { @@ -53,6 +54,11 @@ func (e *Env) startListener(port int, replayPort int) { listener.Settings.Address = "127.0.0.1" listener.Settings.ReplayAddress = "127.0.0.1:" + strconv.Itoa(replayPort) listener.Settings.Port = port + + if e.ListenerLimit != 0 { + listener.Settings.ReplayAddress += "|" + strconv.Itoa(e.ListenerLimit) + } + listener.Run() } @@ -127,7 +133,7 @@ func TestReplay(t *testing.T) { } } -func rateLimitEnv(limit int) int32 { +func rateLimitEnv(replayLimit int, listenerLimit int) int32 { var processed int32 listenHandler := func(w http.ResponseWriter, r *http.Request) { @@ -142,7 +148,8 @@ func rateLimitEnv(limit int) int32 { env := &Env{ ListenHandler: listenHandler, ReplayHandler: replayHandler, - ReplayLimit: limit, + ReplayLimit: replayLimit, + ListenerLimit: listenerLimit, } p := env.start() @@ -158,7 +165,7 @@ func rateLimitEnv(limit int) int32 { } func TestWithoutReplayRateLimit(t *testing.T) { - processed := rateLimitEnv(0) + processed := rateLimitEnv(0, 0) if processed != 10 { t.Error("It should forward all requests without rate-limiting", processed) @@ -166,9 +173,17 @@ func TestWithoutReplayRateLimit(t *testing.T) { } func TestReplayRateLimit(t *testing.T) { - processed := rateLimitEnv(5) + processed := rateLimitEnv(5, 0) if processed != 5 { t.Error("It should forward only 5 requests with rate-limiting", processed) } } + +func TestListenerRateLimit(t *testing.T) { + processed := rateLimitEnv(0, 3) + + if processed != 3 { + t.Error("It should forward only 3 requests with rate-limiting", processed) + } +} diff --git a/listener/listener.go b/listener/listener.go index 2733563..d81b957 100644 --- a/listener/listener.go +++ b/listener/listener.go @@ -13,6 +13,7 @@ import ( "net/http" "os" "strconv" + "time" ) // Enable debug logging only if "--verbose" flag passed @@ -24,7 +25,7 @@ func Debug(v ...interface{}) { func ReplayServer() (conn net.Conn, err error) { // Connection to reaplay server - conn, err = net.Dial("tcp", Settings.ReplayAddress) + conn, err = net.Dial("tcp", Settings.ReplayServer()) if err != nil { log.Println("Connection error ", err, Settings.ReplayAddress) @@ -42,15 +43,33 @@ func Run() { } fmt.Println("Listening for HTTP traffic on", Settings.Address+":"+strconv.Itoa(Settings.Port)) - fmt.Println("Forwarding requests to replay server:", Settings.ReplayAddress) + fmt.Println("Forwarding requests to replay server:", Settings.ReplayServer(), "Limit:", Settings.ReplayLimit) // Sniffing traffic from given address listener := RAWTCPListen(Settings.Address, Settings.Port) + currentTime := time.Now().UnixNano() + currentRPS := 0 + for { // Receiving TCPMessage object m := listener.Receive() + if Settings.ReplayLimit != 0 { + if (time.Now().UnixNano() - currentTime) > time.Second.Nanoseconds() { + currentTime = time.Now().UnixNano() + currentRPS = 0 + } + + if currentRPS >= Settings.ReplayLimit { + break + } + + currentRPS++ + } + + fmt.Println(currentRPS, Settings.ReplayLimit) + go sendMessage(m) } } diff --git a/listener/settings.go b/listener/settings.go index 840dbfd..4d6cd1c 100644 --- a/listener/settings.go +++ b/listener/settings.go @@ -3,6 +3,8 @@ package listener import ( "flag" "os" + "strconv" + "strings" ) const ( @@ -18,11 +20,23 @@ type ListenerSettings struct { ReplayAddress string + ReplayLimit int + Verbose bool } var Settings ListenerSettings = ListenerSettings{} +func (s *ListenerSettings) ReplayServer() string { + host_info := strings.Split(s.ReplayAddress, "|") + + if len(host_info) > 1 { + s.ReplayLimit, _ = strconv.Atoi(host_info[1]) + } + + return host_info[0] +} + func init() { if len(os.Args) < 2 || os.Args[1] != "listen" { return From 3c84a0c0d94499e2fe74e03ed1782a10014f80cf Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Mon, 5 Aug 2013 20:48:01 +0400 Subject: [PATCH 6/7] Remove debugging --- listener/listener.go | 2 -- 1 file changed, 2 deletions(-) diff --git a/listener/listener.go b/listener/listener.go index d81b957..221ea14 100644 --- a/listener/listener.go +++ b/listener/listener.go @@ -68,8 +68,6 @@ func Run() { currentRPS++ } - fmt.Println(currentRPS, Settings.ReplayLimit) - go sendMessage(m) } } From 09175eade06d46724cdd96416ee9069b5bc68a5b Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Mon, 5 Aug 2013 20:51:23 +0400 Subject: [PATCH 7/7] Updated README --- README.md | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 7faad15..e91947b 100644 --- a/README.md +++ b/README.md @@ -32,7 +32,7 @@ gor replay -f http://staging.server -p 28020 ## Advanced use ### Rate limiting -The replay server supports rate limiting. It can be useful if you want +Both replay and listener supports rate limiting. It can be useful if you want forward only part of production traffic and not overload your staging environment. You can specify your desired requests per second using the "|" operator after the server address: @@ -42,6 +42,12 @@ environment. You can specify your desired requests per second using the gor replay -f "http://staging.server|10" ``` +``` +# replay server will not get more than 10 requests per second +# useful for high-load environments +gor listen -p 8080 -r "replay.server.local:28020|10" +``` + ### Forward to multiple addresses You can forward traffic to multiple endpoints. Just separate the addresses by comma.