From a1f9cc501b207ec8a4d9853bac5b4842aa7fcb32 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fernando=20=C3=81lvarez?= Date: Mon, 2 Sep 2013 22:16:07 +0200 Subject: [PATCH] General review: - Fix typos - Follow some Go good practices - Better documentation, more consistent godoc.org - Refactoring --- CHANGELOG.md | 6 +++--- gor.go | 18 +++++++++--------- integration_test.go | 16 +++++++--------- listener/listener.go | 13 +++++++------ listener/listener_test.go | 2 +- listener/raw_tcp_listener.go | 6 ++++-- listener/settings.go | 14 ++++++++------ listener/tcp_message.go | 12 +++++++----- listener/tcp_packet.go | 26 ++++++++++++-------------- replay/replay.go | 11 ++++++----- replay/request_factory.go | 12 +++++++----- replay/request_stats.go | 18 +++++++++--------- replay/settings.go | 18 +++++++++++------- 13 files changed, 91 insertions(+), 81 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index eda6098..8c27605 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,5 @@ v0.3.5 - 15 Sep 2013 -* Significally improved test coverage +* Significantly improved test coverage * Fixed bug with redirect replay https://github.com/buger/gor/pull/15 * Added limit on listener side * Improved stability (catch and log panic, instead of exiting) @@ -7,8 +7,8 @@ v0.3.5 - 15 Sep 2013 v0.3.3 - 22 Jun 2013 * Using TCP instead of UDP for communication between Listener and Replay -* Significally improved performance -* Fixed bugs causing locking and message droping (concurrency issues) +* Significantly improved performance +* Fixed bugs causing locking and message dropping (concurrency issues) * Rewrote concurrency model to use more channels v0.3 - 10 Jun 2013 diff --git a/gor.go b/gor.go index fb14b49..9bf6d78 100644 --- a/gor.go +++ b/gor.go @@ -8,27 +8,29 @@ package main import ( "flag" "fmt" - "github.com/buger/gor/listener" - "github.com/buger/gor/replay" "log" "os" "runtime/pprof" "time" + + "github.com/buger/gor/listener" + "github.com/buger/gor/replay" ) const ( VERSION = "0.3.5" ) -var cpuprofile = flag.String("cpuprofile", "", "write cpu profile to file") -var memprofile = flag.String("memprofile", "", "write memory profile to this file") +var ( + mode string + cpuprofile = flag.String("cpuprofile", "", "write cpu profile to file") + memprofile = flag.String("memprofile", "", "write memory profile to this file") +) func main() { defer func() { if r := recover(); r != nil { - var ok bool - _, ok = r.(error) - if !ok { + if _, ok := r.(error); !ok { fmt.Errorf("pkg: %v", r) } } @@ -36,8 +38,6 @@ func main() { fmt.Println("Version:", VERSION) - mode := "unknown" - if len(os.Args) > 1 { mode = os.Args[1] } diff --git a/integration_test.go b/integration_test.go index 24b8d78..4086371 100644 --- a/integration_test.go +++ b/integration_test.go @@ -1,18 +1,15 @@ package main import ( - "testing" - - "github.com/buger/gor/listener" - "github.com/buger/gor/replay" - - "time" - "fmt" "net/http" "strconv" - "sync/atomic" + "testing" + "time" + + "github.com/buger/gor/listener" + "github.com/buger/gor/replay" ) func isEqual(t *testing.T, a interface{}, b interface{}) { @@ -56,7 +53,7 @@ func (e *Env) startListener(port int, replayPort int) { listener.Settings.Port = port if e.ListenerLimit != 0 { - listener.Settings.ReplayAddress += "|" + strconv.Itoa(e.ListenerLimit) + listener.Settings.ReplayLimit = e.ListenerLimit } listener.Run() @@ -65,6 +62,7 @@ func (e *Env) startListener(port int, replayPort int) { func (e *Env) startReplay(port int, forwardPort int) { replay.Settings.Verbose = e.Verbose replay.Settings.Host = "127.0.0.1" + replay.Settings.Address = "127.0.0.1:" + strconv.Itoa(port) replay.Settings.ForwardAddress = "127.0.0.1:" + strconv.Itoa(forwardPort) replay.Settings.Port = port diff --git a/listener/listener.go b/listener/listener.go index deeb671..77418e3 100644 --- a/listener/listener.go +++ b/listener/listener.go @@ -1,7 +1,7 @@ // Listener capture TCP traffic using RAW SOCKETS. // Note: it requires sudo or root access. // -// Rigt now it suport only HTTP +// Right now it supports only HTTP package listener import ( @@ -16,16 +16,17 @@ import ( "time" ) -// Enable debug logging only if "--verbose" flag passed +// Debug enables logging only if "--verbose" flag passed func Debug(v ...interface{}) { if Settings.Verbose { log.Println(v...) } } +// ReplayServer returns a connection to the replay server and error if some func ReplayServer() (conn net.Conn, err error) { - // Connection to reaplay server - conn, err = net.Dial("tcp", Settings.ReplayServer()) + // Connection to replay server + conn, err = net.Dial("tcp", Settings.ReplayAddress) if err != nil { log.Println("Connection error ", err, Settings.ReplayAddress) @@ -34,7 +35,7 @@ func ReplayServer() (conn net.Conn, err error) { return } -// Because its sub-program, Run acts as `main` +// Run acts as `main` function of a listener func Run() { if os.Getuid() != 0 { fmt.Println("Please start the listener as root or sudo!") @@ -43,7 +44,7 @@ func Run() { } fmt.Println("Listening for HTTP traffic on", Settings.Address+":"+strconv.Itoa(Settings.Port)) - fmt.Println("Forwarding requests to replay server:", Settings.ReplayServer(), "Limit:", Settings.ReplayLimit) + fmt.Println("Forwarding requests to replay server:", Settings.ReplayAddress, "Limit:", Settings.ReplayLimit) // Sniffing traffic from given address listener := RAWTCPListen(Settings.Address, Settings.Port) diff --git a/listener/listener_test.go b/listener/listener_test.go index 319ed55..775399c 100644 --- a/listener/listener_test.go +++ b/listener/listener_test.go @@ -40,6 +40,6 @@ func TestSendMessage(t *testing.T) { buf = buf[0:n] if bytes.Compare(buf, msg.Bytes()) != 0 { - t.Errorf("Original and reveived requests does not match") + t.Errorf("Original and received requests does not match") } } diff --git a/listener/raw_tcp_listener.go b/listener/raw_tcp_listener.go index ce2529b..911c83c 100644 --- a/listener/raw_tcp_listener.go +++ b/listener/raw_tcp_listener.go @@ -25,6 +25,7 @@ type RAWTCPListener struct { port int // Port to listen } +// RAWTCPListen creates a listener to capture traffic from RAW_SOCKET func RAWTCPListen(addr string, port int) (listener *RAWTCPListener) { listener = &RAWTCPListener{} @@ -49,7 +50,7 @@ func (t *RAWTCPListener) listen() { t.c_messages <- message t.deleteMessage(message) - // We need to use channgels to process each packet to avoid data races + // We need to use channels to process each packet to avoid data races case packet := <-t.c_packets: t.processTCPPacket(packet) } @@ -58,7 +59,7 @@ func (t *RAWTCPListener) listen() { // Deleting messages that came from t.c_del_message channel func (t *RAWTCPListener) deleteMessage(message *TCPMessage) bool { - var idx int = -1 + idx := -1 // Searching for given message in messages buffer for i, m := range t.messages { @@ -154,6 +155,7 @@ func (t *RAWTCPListener) processTCPPacket(packet *TCPPacket) { message.c_packets <- packet } +// Receive TCP messages from the listener channel func (t *RAWTCPListener) Receive() *TCPMessage { return <-t.c_messages } diff --git a/listener/settings.go b/listener/settings.go index 4d6cd1c..fe2b9c9 100644 --- a/listener/settings.go +++ b/listener/settings.go @@ -14,6 +14,7 @@ const ( defaultReplayAddress = "localhost:28020" ) +// ListenerSettings contain all the needed configuration for setting up the listener type ListenerSettings struct { Port int Address string @@ -27,14 +28,15 @@ type ListenerSettings struct { var Settings ListenerSettings = ListenerSettings{} -func (s *ListenerSettings) ReplayServer() string { - host_info := strings.Split(s.ReplayAddress, "|") +// ReplayServer generates ReplayLimit and ReplayAddress settings out of the replayAddress +func (s *ListenerSettings) ReplayServer(replayAddress string) { + host_info := strings.Split(replayAddress, "|") if len(host_info) > 1 { s.ReplayLimit, _ = strconv.Atoi(host_info[1]) } - return host_info[0] + s.ReplayAddress = host_info[0] } func init() { @@ -43,10 +45,10 @@ func init() { } flag.IntVar(&Settings.Port, "p", defaultPort, "Specify the http server port whose traffic you want to capture") + flag.StringVar(&Settings.Address, "ip", defaultAddress, "Specify IP address to listen") - flag.StringVar(&Settings.Address, "ip", defaultAddress, "Specifi IP address to listen") - - flag.StringVar(&Settings.ReplayAddress, "r", defaultReplayAddress, "Address of replay server.") + replayAddress := flag.String("r", defaultReplayAddress, "Address of replay server.") + Settings.ReplayServer(*replayAddress) flag.BoolVar(&Settings.Verbose, "verbose", false, "Log requests") } diff --git a/listener/tcp_message.go b/listener/tcp_message.go index b34ec8c..0172149 100644 --- a/listener/tcp_message.go +++ b/listener/tcp_message.go @@ -10,9 +10,9 @@ const MSG_EXPIRE = 200 * time.Millisecond // TCPMessage ensure that all TCP packets for given request is received, and processed in right sequence // Its needed because all TCP message can be fragmented or re-transmitted // -// Each TCP Packet have 2 ids: acknowledgement - message_id, and sequence - packet_id +// Each TCP Packet have 2 ids: acknowledgment - message_id, and sequence - packet_id // Message can be compiled from unique packets with same message_id which sorted by sequence -// Message is received if we did't receive any packets for 200ms +// Message is received if we didn't receive any packets for 200ms type TCPMessage struct { Ack uint32 // Message ID packets []*TCPPacket @@ -27,6 +27,7 @@ type TCPMessage struct { c_del_message chan *TCPMessage } +// NewTCPMessage pointer created from a Acknowledgment number and a channel of messages readuy to be deleted func NewTCPMessage(Ack uint32, c_del chan *TCPMessage) (msg *TCPMessage) { msg = &TCPMessage{Ack: Ack} @@ -54,12 +55,13 @@ func (t *TCPMessage) listen() { } } +// Timeout notifies message to stop listening, close channel and message ready to be sent func (t *TCPMessage) Timeout() { t.c_closing <- 1 // Notify to stop listen loop and close channel - t.c_del_message <- t // Notify RAWListener that message is ready to be send to replay server + t.c_del_message <- t // Notify RAWListener that message is ready to be sent to replay server } -// Sort packets in right orders and return message content +// Bytes sorts packets in right orders and return message content func (t *TCPMessage) Bytes() (output []byte) { mk := make([]int, len(t.packets)) @@ -78,7 +80,7 @@ func (t *TCPMessage) Bytes() (output []byte) { return } -// Add packet to the message and ensure packet uniquiness +// AddPacket to the message and ensure packet uniqueness // TCP allows that packet can be re-send multiple times func (t *TCPMessage) AddPacket(packet *TCPPacket) { if t.expired { diff --git a/listener/tcp_packet.go b/listener/tcp_packet.go index e466d6e..0020279 100644 --- a/listener/tcp_packet.go +++ b/listener/tcp_packet.go @@ -36,43 +36,41 @@ type TCPPacket struct { Data []byte } -func NewTCPPacket(b []byte) (p *TCPPacket) { - p = &TCPPacket{Data: b} - p.ParseFast() +// NewTCPPacket pointer from data in []byte +func NewTCPPacket(b []byte) (t *TCPPacket) { + t = &TCPPacket{Data: b} + t.ParseBasic() - return p + return t } -// Inspired by: https://github.com/miekg/pcap/blob/master/packet.go +// Parse TCP Packet, inspired by: https://github.com/miekg/pcap/blob/master/packet.go func (t *TCPPacket) Parse() { + t.ParseBasic() t.SrcPort = binary.BigEndian.Uint16(t.Data[0:2]) t.DestPort = binary.BigEndian.Uint16(t.Data[2:4]) - t.Seq = binary.BigEndian.Uint32(t.Data[4:8]) - t.Ack = binary.BigEndian.Uint32(t.Data[8:12]) - t.DataOffset = (t.Data[12] & 0xF0) >> 4 t.Flags = binary.BigEndian.Uint16(t.Data[12:14]) & 0x1FF t.Window = binary.BigEndian.Uint16(t.Data[14:16]) t.Checksum = binary.BigEndian.Uint16(t.Data[16:18]) t.Urgent = binary.BigEndian.Uint16(t.Data[18:20]) - - t.Data = t.Data[t.DataOffset*4:] } -// Parse only needed set of fields -func (t *TCPPacket) ParseFast() { +// ParseBasic set of fields +func (t *TCPPacket) ParseBasic() { t.Seq = binary.BigEndian.Uint32(t.Data[4:8]) t.Ack = binary.BigEndian.Uint32(t.Data[8:12]) - t.DataOffset = (t.Data[12] & 0xF0) >> 4 + t.Data = t.Data[t.DataOffset*4:] } +// String output for a TCP Packet func (t *TCPPacket) String() string { return strings.Join([]string{ "Source port: " + strconv.Itoa(int(t.SrcPort)), "Dest port:" + strconv.Itoa(int(t.DestPort)), "Sequence:" + strconv.Itoa(int(t.Seq)), - "Acknowledgement:" + strconv.Itoa(int(t.Ack)), + "Acknowledgment:" + strconv.Itoa(int(t.Ack)), "Header len:" + strconv.Itoa(int(t.DataOffset)), "Flag ns:" + strconv.FormatBool(t.Flags&TCP_NS != 0), diff --git a/replay/replay.go b/replay/replay.go index efb487b..cd2c978 100644 --- a/replay/replay.go +++ b/replay/replay.go @@ -35,13 +35,14 @@ import ( const bufSize = 4096 -// Enable debug logging only if "--verbose" flag passed +// Debug enables logging only if "--verbose" flag passed func Debug(v ...interface{}) { if Settings.Verbose { log.Println(v...) } } +// ParseRequest in []byte returns a http request or an error func ParseRequest(data []byte) (request *http.Request, err error) { buf := bytes.NewBuffer(data) reader := bufio.NewReader(buf) @@ -50,13 +51,13 @@ func ParseRequest(data []byte) (request *http.Request, err error) { return } -// Because its sub-program, Run acts as `main` +// Run acts as `main` function of replay // Replay server listen to UDP traffic from Listeners // Each request processed by RequestFactory func Run() { - listener, err := net.Listen("tcp", Settings.Address()) + listener, err := net.Listen("tcp", Settings.Address) - log.Println("Starting replay server at:", Settings.Address()) + log.Println("Starting replay server at:", Settings.Address) if err != nil { log.Fatal("Can't start:", err) @@ -97,7 +98,7 @@ func handleConnection(conn net.Conn, rf *RequestFactory) error { case io.EOF: read = false case nil: - response = append(response, buf[0:n]...) + response = append(response, buf[:n]...) if n < bufSize { read = false } diff --git a/replay/request_factory.go b/replay/request_factory.go index 1d8e65f..8ebb592 100644 --- a/replay/request_factory.go +++ b/replay/request_factory.go @@ -6,6 +6,8 @@ import ( "net/url" ) +// HttpResponse contains a host, a http request, +// a http response and an error type HttpResponse struct { host *ForwardHost req *http.Request @@ -13,7 +15,7 @@ type HttpResponse struct { err error } -// Class for processing requests +// RequestFactory processes requests // // Basic workflow: // @@ -26,7 +28,7 @@ type RequestFactory struct { c_requests chan *http.Request } -// RequestFactory contstuctor +// NewRequestFactory returns a RequestFactory pointer // One created, it starts listening for incoming requests: requests channel func NewRequestFactory() (factory *RequestFactory) { factory = &RequestFactory{} @@ -38,7 +40,7 @@ func NewRequestFactory() (factory *RequestFactory) { return } -// Disable redirects https://github.com/buger/gor/pull/15 +// customCheckRedirect disables redirects https://github.com/buger/gor/pull/15 func customCheckRedirect(req *http.Request, via []*http.Request) error { if len(via) >= 0 { return errors.New("stopped after 2 redirects") @@ -46,7 +48,7 @@ func customCheckRedirect(req *http.Request, via []*http.Request) error { return nil } -// Forward http request to given host +// sendRequest forwards http request to a given host func (f *RequestFactory) sendRequest(host *ForwardHost, request *http.Request) { client := &http.Client{ CheckRedirect: customCheckRedirect, @@ -71,7 +73,7 @@ func (f *RequestFactory) sendRequest(host *ForwardHost, request *http.Request) { f.c_responses <- &HttpResponse{host, request, resp, err} } -// Handle incoming requests, and they responses +// handleRequests and their responses func (f *RequestFactory) handleRequests() { hosts := Settings.ForwardedHosts() diff --git a/replay/request_stats.go b/replay/request_stats.go index d1bbf32..cf26282 100644 --- a/replay/request_stats.go +++ b/replay/request_stats.go @@ -4,31 +4,31 @@ import ( "time" ) -// Stats stores in context of current timestamp +// RequestStat stores in context of current timestamp type RequestStat struct { timestamp int64 Codes map[int]int // { 200: 10, 404:2, 500:1 } Count int // All requests including errors - Errors int // Rquests with errors (timeout or host not reachable). Not include 50x errors. + Errors int // Requests with errors (timeout or host not reachable). Not include 50x errors. host *ForwardHost } -// Ensure that current stats is actual (for current timestamp) +// Touch ensures that current stats is actual (for current timestamp) func (s *RequestStat) Touch() { if s.timestamp != time.Now().Unix() { s.reset() } } -// Called on request start +// IncReq is called on request start func (s *RequestStat) IncReq() { s.Count++ } -// Called after response +// IncResp is called after response func (s *RequestStat) IncResp(resp *HttpResponse) { s.Touch() @@ -40,11 +40,11 @@ func (s *RequestStat) IncResp(resp *HttpResponse) { s.Codes[resp.resp.StatusCode]++ } -// Updated stats timestamp to current time and reset to zero all stats values +// reset updates 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 { - Debug("Host:", s.host.Url, "Requests:", s.Count, "Errors:", s.Errors, "Status codes:", s.Codes) + if s.timestamp != 0 { + Debug("Host:", s.host.Url, "Requests:", s.Count, "Errors:", s.Errors, "Status codes:", s.Codes) } s.timestamp = time.Now().Unix() @@ -54,7 +54,7 @@ func (s *RequestStat) reset() { s.Errors = 0 } -// RequestStat constructor +// NewRequestStats returns a RequestStat pointer func NewRequestStats(host *ForwardHost) (stat *RequestStat) { stat = &RequestStat{host: host} stat.reset() diff --git a/replay/settings.go b/replay/settings.go index 8ac8dfb..3aa3b3c 100644 --- a/replay/settings.go +++ b/replay/settings.go @@ -7,6 +7,7 @@ import ( "strings" ) +// ForwardHost where to forward requests type ForwardHost struct { Url string Limit int @@ -14,15 +15,19 @@ type ForwardHost struct { Stat *RequestStat } +// ReplaySettings ListenerSettings contain all the needed configuration for setting up the replay type ReplaySettings struct { Port int Host string + Address string ForwardAddress string Verbose bool } +var Settings ReplaySettings = ReplaySettings{} + // 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" @@ -50,13 +55,11 @@ 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) +// SetAddress with port, e.g.: 127.0.0.1:28020 +func (r *ReplaySettings) SetAddress() { + r.Address = r.Host + ":" + strconv.Itoa(r.Port) } -var Settings ReplaySettings = ReplaySettings{} - func init() { if len(os.Args) < 2 || os.Args[1] != "replay" { return @@ -66,14 +69,15 @@ func init() { defaultPort = 28020 defaultHost = "0.0.0.0" - defaultAddress = "http://localhost:8080" + defaultForwardAddress = "http://localhost:8080" ) flag.IntVar(&Settings.Port, "p", defaultPort, "specify port number") flag.StringVar(&Settings.Host, "ip", defaultHost, "ip addresses to listen on") - 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") + Settings.SetAddress() + flag.StringVar(&Settings.ForwardAddress, "f", defaultForwardAddress, "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.BoolVar(&Settings.Verbose, "verbose", false, "Log requests") }