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

Merge branch 'master' into refactoring

Browse Source 
Conflicts:
	listener/raw_tcp_listener.go
	listener/tcp_message.go
	listener/tcp_packet.go
This commit is contained in:
Leonid Bugaev
2013-09-30 11:26:35 +02:00
parent f07586a12e 8535872b10
commit 6946fc6036
14 changed files with 123 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+10 -3
View File
@@ -1,11 +1,18 @@
v0.3.5 - 15 Sep 2013
* 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)
* Added License file
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
* Use RAW_SOCKETS instead of tcpdump
* Own TCP stack
* All HTTP request types support
* Simplified request parsing
* Simplified request parsing
README.md
+5 -8
View File
@@ -74,9 +74,9 @@ Usage of ./bin/gor-linux:
-p=28020: specify port number
```
## Pre-build binaries
## Latest releases (including binaries)
[Download binaries (linux 32/64, darwin)](https://drive.google.com/folderview?id=0B46uay48NwcfWFowc1E4a1BISVU&usp=sharing)
https://github.com/buger/gor/releases
## Building from source
1. Setup standard Go environment http://golang.org/doc/code.html and ensure that $GOPATH environment variable properly set.
@@ -104,10 +104,7 @@ Yes. ~~Right now it supports only "GET" requests.~~
4. Push to the branch (git push origin my-new-feature)
5. Create new Pull Request
## TODO
## Companies using Gor
Use buffering for request throttling instead of simple rate limiting.
Better stats
Optimize for load testing cases
* http://granify.com
* To add your company drop me a line to github.com/buger or leonsbox@gmail.com
gor.go
+9 -9
View File
@@ -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]
}
integration_test.go
+7 -9
View File
@@ -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
listener/listener.go
+7 -6
View File
@@ -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)
listener/listener_test.go
+1 -1
View File
@@ -42,7 +42,7 @@ 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")
}
}
listener/raw_tcp_listener.go
+29 -1
View File
@@ -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{}
@@ -50,13 +51,39 @@ func (t *RAWTCPListener) listen() {
t.c_messages <- message
delete(t.messages, message.Ack)
// 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)
}
}
}
// Deleting messages that came from t.c_del_message channel
func (t *RAWTCPListener) deleteMessage(message *TCPMessage) bool {
idx := -1
// Searching for given message in messages buffer
for i, m := range t.messages {
if m.Ack == message.Ack {
idx = i
break
}
}
if idx == -1 {
return false
}
// Delete element from array
// Note: that this version for arrays that consist of pointers
// https://code.google.com/p/go-wiki/wiki/SliceTricks
copy(t.messages[idx:], t.messages[idx+1:])
t.messages[len(t.messages)-1] = nil // Ensure that value will be garbage-collected.
t.messages = t.messages[:len(t.messages)-1]
return true
}
func (t *RAWTCPListener) readRAWSocket() {
conn, e := net.ListenPacket("ip4:tcp", t.addr)
defer conn.Close()
@@ -130,6 +157,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
}
listener/settings.go
+8 -6
View File
@@ -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")
}
listener/tcp_message.go
+6 -4
View File
@@ -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
@@ -24,6 +24,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}
@@ -52,12 +53,13 @@ func (t *TCPMessage) listen() {
}
}
// Timeout notifies message to stop listening, close channel and message ready to be sent
func (t *TCPMessage) Timeout() {
close(t.c_packets) // 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
}
// 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))
@@ -76,7 +78,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) {
packetFound := false
listener/tcp_packet.go
+8 -11
View File
@@ -40,39 +40,36 @@ func ParseTCPPacket(b []byte) (p *TCPPacket) {
p = &TCPPacket{Data: b}
p.ParseFast()
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),
replay/replay.go
+6 -5
View File
@@ -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
}
replay/request_factory.go
+7 -5
View File
@@ -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()
replay/request_stats.go
+9 -9
View File
@@ -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()
replay/settings.go
+11 -7
View File
@@ -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")
}