From 5647c2d1756efeaf92ff3a57d07c0adb120be401 Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Sun, 2 Jun 2013 11:17:07 +0600 Subject: [PATCH] Updated documentation --- gor.go | 57 ++++++++++++++++++------------- listener/listener.go | 72 +++++++++++++++++++++++---------------- replay/replay.go | 36 +++++++++++++++++--- replay/request_factory.go | 34 +++++++++++++----- replay/request_stats.go | 63 ++++++++++++++++++---------------- replay/settings.go | 10 +++--- 6 files changed, 174 insertions(+), 98 deletions(-) diff --git a/gor.go b/gor.go index 1ace6cb..9df00f5 100644 --- a/gor.go +++ b/gor.go @@ -1,35 +1,46 @@ +// Gor is simple http traffic replication tool written in Go. Its main goal to replay traffic from production servers to staging and dev environments. +// Now you can test your code on real user sessions in an automated and repeatable fashion. +// +// Gor consists of 2 parts: listener and replay servers. +// Listener catch http traffic from given port in real-time and send it to replay server via UDP. Replay server forwards traffic to given address. package main import ( - "flag" - "fmt" - "github.com/buger/gor/listener" - "github.com/buger/gor/replay" - "os" + "flag" + "fmt" + "github.com/buger/gor/listener" + "github.com/buger/gor/replay" + "os" +) + +const ( + VERSION = 0.1 ) func main() { - mode := "unknown" + fmt.Println("Version:", VERSION) - if len(os.Args) > 1 { - mode = os.Args[1] - } + mode := "unknown" - if mode != "listen" && mode != "replay" { - fmt.Println("Usage: \n\tgor listen -h\n\tgor replay -h") - return - } - - // Remove mode attr - os.Args = append(os.Args[:1], os.Args[2:]...) + if len(os.Args) > 1 { + mode = os.Args[1] + } - flag.Parse() + if mode != "listen" && mode != "replay" { + fmt.Println("Usage: \n\tgor listen -h\n\tgor replay -h") + return + } - switch mode { - case "listen": - listener.Run() - case "replay": - replay.Run() - } + // Remove mode attr + os.Args = append(os.Args[:1], os.Args[2:]...) + + flag.Parse() + + switch mode { + case "listen": + listener.Run() + case "replay": + replay.Run() + } } diff --git a/listener/listener.go b/listener/listener.go index ebe53e8..89a4448 100644 --- a/listener/listener.go +++ b/listener/listener.go @@ -1,18 +1,20 @@ +// Listener capture TCP traffic right from given port using `tcpdump` utility. +// Note: it requires sudo or root access. +// +// Rigt now it suport only HTTP, and only GET requests. package listener import ( - "fmt" - "io" - "log" - "os/exec" - //"time" "bufio" "bytes" "encoding/gob" - //"errors" "flag" + "fmt" + "io" + "log" "net" "os" + "os/exec" "regexp" "strconv" "strings" @@ -28,14 +30,16 @@ type ListenerSettings struct { var settings ListenerSettings = ListenerSettings{} type HttpRequest struct { - Tag string - Method string - Url string - Headers map[string]string + Tag string // Not used yet + Method string // Right now only 'GET' + Url string // Request URL + Headers map[string]string // Request Headers } -func readOutput(pipe io.ReadCloser, c chan *HttpRequest, err chan int) { - re := regexp.MustCompile("(GET) (/.*) HTTP/1.1") +// Parse `tcpdump` output to find HTTP GET requests +// When HttpRequest found it get send to `requests` channel +func ParseRequest(pipe io.ReadCloser, requests chan *HttpRequest) { + request_re := regexp.MustCompile("(GET) (/.*) HTTP/1.1") headers_re := regexp.MustCompile("([^ ]*): (.*)") reader := bufio.NewScanner(pipe) @@ -47,8 +51,11 @@ func readOutput(pipe io.ReadCloser, c chan *HttpRequest, err chan int) { for reader.Scan() { line := reader.Text() + // HTTP/1.1 match finds both requests and response + // Index is used instead of Regexp just for speed if strings.Index(line, "HTTP/1.1") != -1 { - match := re.FindAllString(line, -1) + // Allow only requests + match := request_re.FindAllString(line, -1) if len(match) > 0 { info := strings.Split(match[0], " ") @@ -64,10 +71,17 @@ func readOutput(pipe io.ReadCloser, c chan *HttpRequest, err chan int) { } if requestStarted { + // We assume that empty line is end of request info + // This is true only for GET requests if line == "" { - c <- request + requests <- request requestStarted = false } else { + // All headers comes in this format: + // + // User-Agent: Mozilla + // Content-Type: text/html + // match := headers_re.FindAllString(line, -1) if len(match) > 0 { @@ -80,7 +94,9 @@ func readOutput(pipe io.ReadCloser, c chan *HttpRequest, err chan int) { } } -func sendOutput(c chan *HttpRequest, quite chan int) { +// Sends request to replay server via UDP +// Before sending it encode request object using standard gob encoder +func SendRequest(requests chan *HttpRequest) { serverAddr, err := net.ResolveUDPAddr("udp4", settings.replayAddress) conn, err := net.DialUDP("udp", nil, serverAddr) @@ -92,7 +108,7 @@ func sendOutput(c chan *HttpRequest, quite chan int) { for { select { - case request := <-c: + case request := <-requests: fmt.Println("Request:", request.Url) msg := bytes.Buffer{} @@ -105,27 +121,24 @@ func sendOutput(c chan *HttpRequest, quite chan int) { if err != nil { log.Println("encode error:", err) } - - case <-quite: - conn.Close() - return } } } +// Because its sub-program, Run acts as `main` func Run() { - if os.Getuid() != 0 { - fmt.Println("Please start the listener as root or sudo!") - fmt.Println("This is required since listener sniff traffic on given port.") - os.Exit(1) - } + if os.Getuid() != 0 { + fmt.Println("Please start the listener as root or sudo!") + fmt.Println("This is required since listener sniff traffic on given port.") + os.Exit(1) + } if !strings.Contains(settings.replayAddress, ":") { settings.replayAddress = settings.replayAddress + ":28020" } + // TODO: use RAW_SOCKETS instead of tcpdump cmd := exec.Command("tcpdump", "-vv", "-A", "-i", settings.networkInterface, "port "+strconv.Itoa(settings.port)) - //cmd := exec.Command("ls", "-al") stdout, _ := cmd.StdoutPipe() cmd.Stderr = os.Stderr @@ -134,11 +147,10 @@ func Run() { log.Fatal(err) } - c := make(chan *HttpRequest) - err := make(chan int) + requests := make(chan *HttpRequest) - go readOutput(stdout, c, err) - go sendOutput(c, err) + go ParseRequest(stdout, requests) + go SendRequest(requests) if err := cmd.Wait(); err != nil { flag.Usage() diff --git a/replay/replay.go b/replay/replay.go index 64158a0..c872669 100644 --- a/replay/replay.go +++ b/replay/replay.go @@ -1,3 +1,27 @@ +// Replay server receive requests objects from Listeners and forward it to given address. +// Basic usage: +// +// gor replay -f http://staging.server +// +// +// Rate limiting +// +// It can be useful if you want forward only part of production traffic, not to overload staging environment. You can specify desired request per second using "|" operator after server address: +// +// # staging.server not get more than 10 requests per second +// gor replay -f "http://staging.server|10" +// +// +// Forward to multiple addresses +// +// Just separate addresses by coma: +// gor replay -f "http://staging.server|10,http://dev.server|20" +// +// +// For more help run: +// +// gor replay -h +// package replay import ( @@ -10,11 +34,12 @@ import ( "os" ) -var settings ReplaySettings = ReplaySettings{} - const bufSize = 1024 * 10 -func decodeRequest(enc []byte) (request *HttpRequest, err error) { +var settings ReplaySettings = ReplaySettings{} + +// Decode HttpRequest object using standard gob decoder +func DecodeRequest(enc []byte) (request *HttpRequest, err error) { var buf bytes.Buffer buf.Write(enc) @@ -26,6 +51,9 @@ func decodeRequest(enc []byte) (request *HttpRequest, err error) { return } +// Because its sub-program, Run acts as `main` +// Replay server listen to UDP traffic from Listeners +// Each request processed by RequestFactory func Run() { var buf [bufSize]byte @@ -89,5 +117,5 @@ func init() { flag.StringVar(&settings.host, "ip", defaultHost, "ip addresses to listen on") - flag.StringVar(&settings.address, "f", defaultAddress, "http address to forward traffic.\n\tYou can limit requests per second by adding `|num` after address.\n\tIf you have multiple addresses with different limits. For example: http://staging.example.com|100,http://dev.example.com|10") + flag.StringVar(&settings.forwardAddress, "f", defaultAddress, "http address to forward traffic.\n\tYou can limit requests per second by adding `|num` after address.\n\tIf you have multiple addresses with different limits. For example: http://staging.example.com|100,http://dev.example.com|10") } diff --git a/replay/request_factory.go b/replay/request_factory.go index ff4c2b4..0e6cb84 100644 --- a/replay/request_factory.go +++ b/replay/request_factory.go @@ -1,15 +1,16 @@ package replay import ( - "net/http" "fmt" + "net/http" ) +// Userd for transfering Request info between Listener and Replay server type HttpRequest struct { - Tag string - Method string - Url string - Headers map[string]string + Tag string // Not used yet + Method string // Right now only 'GET' + Url string // Request URL + Headers map[string]string // Request Headers } type HttpResponse struct { @@ -19,11 +20,21 @@ type HttpResponse struct { err error } +// Class for processing requests +// +// Basic workflow: +// +// 1. When request added via Add() it get pushed to `responses` chan +// 2. handleRequest() listen for `responses` chan and decide where request should be forwarded, and apply rate-limit if needed +// 3. sendRequest() forwards request and returns response info to `responses` chan +// 4. handleRequest() listen for `response` channel and updates stats type RequestFactory struct { responses chan *HttpResponse requests chan *HttpRequest } +// RequestFactory contstuctor +// One created, it starts listening for incoming requests: requests channel func NewRequestFactory() (factory *RequestFactory) { factory = &RequestFactory{} factory.responses = make(chan *HttpResponse) @@ -34,6 +45,7 @@ func NewRequestFactory() (factory *RequestFactory) { return } +// Forward http request to given host func (f *RequestFactory) sendRequest(host *ForwardHost, request *HttpRequest) { var req *http.Request @@ -41,6 +53,7 @@ func (f *RequestFactory) sendRequest(host *ForwardHost, request *HttpRequest) { req, err := http.NewRequest("GET", host.Url+request.Url, nil) + // Forwarded request should have same headers for key, value := range request.Headers { req.Header.Add(key, value) } @@ -49,21 +62,24 @@ func (f *RequestFactory) sendRequest(host *ForwardHost, request *HttpRequest) { if err == nil { defer resp.Body.Close() - } + } f.responses <- &HttpResponse{host, request, resp, err} } +// Handle incoming requests, and they responses func (f *RequestFactory) handleRequests() { hosts := settings.ForwardedHosts() for { select { - case req := <- f.requests: + case req := <-f.requests: for _, host := range hosts { + // Ensure that we have actual stats for given timestamp host.Stat.Touch() if host.Limit == 0 || host.Stat.Count < host.Limit { + // Increment Stat.Count host.Stat.IncReq() fmt.Println("Sending request") @@ -72,12 +88,14 @@ func (f *RequestFactory) handleRequests() { fmt.Println("Throttling for host:", host.Url, host.Stat.Count, host.Limit) } } - case resp := <- f.responses: + case resp := <-f.responses: + // Increment returned http code stats, and elapsed time resp.host.Stat.IncResp(resp) } } } +// Add request to channel for further processing func (f *RequestFactory) Add(request *HttpRequest) { f.requests <- request } diff --git a/replay/request_stats.go b/replay/request_stats.go index cf01ded..d122f50 100644 --- a/replay/request_stats.go +++ b/replay/request_stats.go @@ -1,59 +1,64 @@ package replay import ( - "log" - "time" + "log" + "time" ) +// Stats stores in context of current timestamp type RequestStat struct { - timestamp int64 + timestamp int64 - Codes map[int]int + Codes map[int]int // { 200: 10, 404:2, 500:1 } - Count int - Errors int + Count int // All requests including errors + Errors int // Rquests with errors (timeout or host not reachable). Not include 50x errors. - host *ForwardHost + host *ForwardHost } +// Ensure that current stats is actual (for current timestamp) func (s *RequestStat) Touch() { - if s.timestamp != time.Now().Unix() { - s.reset() - } + if s.timestamp != time.Now().Unix() { + s.reset() + } } +// Called on request start func (s *RequestStat) IncReq() { - s.Touch() - - s.Count++ + s.Count++ } +// Called after response func (s *RequestStat) IncResp(resp *HttpResponse) { - s.Touch() + s.Touch() - if resp.err != nil { - s.Errors++ - return - } + if resp.err != nil { + s.Errors++ + return + } - s.Codes[resp.resp.StatusCode]++ + s.Codes[resp.resp.StatusCode]++ } +// Updated stats timestamp to current time and reset to zero all stats values +// TODO: Further on reset it should write stats to file func (s *RequestStat) reset() { - if s.timestamp != 0 { - log.Println("Host:", s.host.Url, "Requests:", s.Count, "Errors:", s.Errors, "Status codes:", s.Codes) - } + if s.timestamp != 0 { + log.Println("Host:", s.host.Url, "Requests:", s.Count, "Errors:", s.Errors, "Status codes:", s.Codes) + } - s.timestamp = time.Now().Unix() + s.timestamp = time.Now().Unix() - s.Codes = make(map[int]int) - s.Count = 0 - s.Errors = 0 + s.Codes = make(map[int]int) + s.Count = 0 + s.Errors = 0 } +// RequestStat constructor func NewRequestStats(host *ForwardHost) (stat *RequestStat) { - stat = &RequestStat{host: host} - stat.reset() + stat = &RequestStat{host: host} + stat.reset() - return + return } diff --git a/replay/settings.go b/replay/settings.go index cb99e83..83e4741 100644 --- a/replay/settings.go +++ b/replay/settings.go @@ -5,7 +5,6 @@ import ( "strings" ) - type ForwardHost struct { Url string Limit int @@ -17,11 +16,13 @@ type ReplaySettings struct { port int host string - limit int - - address string + forwardAddress string } +// ForwardedHosts implements forwardAddress syntax support for multiple hosts (coma separated), and rate limiting by specifing "|maxRps" after host name. +// +// -f "host1,http://host2|10,host3" +// func (r *ReplaySettings) ForwardedHosts() (hosts []*ForwardHost) { hosts = make([]*ForwardHost, 0, 10) @@ -45,6 +46,7 @@ func (r *ReplaySettings) ForwardedHosts() (hosts []*ForwardHost) { return } +// Helper to return address with port, e.g.: 127.0.0.1:28020 func (r *ReplaySettings) Address() string { return r.host + ":" + strconv.Itoa(r.port) }