Updated documentation

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()
+	}
 
 }

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()

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")
 }

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
 }

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
 }

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)
 }