mirror of
https://github.com/buger/goreplay.git
synced 2026-10-21 03:57:58 +00:00
Merge branch 'master' into save-request-to-reply-later
Conflicts: listener/raw_tcp_listener.go replay/replay.go replay/settings.go
This commit is contained in:
Sławosz
+3
-3
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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]
|
||||
}
|
||||
|
||||
+7
-9
@@ -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{}) {
|
||||
@@ -57,7 +54,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()
|
||||
@@ -66,6 +63,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
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
+47
-11
@@ -5,34 +5,36 @@ import (
|
||||
"fmt"
|
||||
"net"
|
||||
"testing"
|
||||
//"time"
|
||||
)
|
||||
|
||||
func getTCPMessage() (msg *TCPMessage) {
|
||||
packet := &TCPPacket{Data: []byte("GET /pub/WWW/ HTTP/1.1\nHost: www.w3.org\r\n\r\n")}
|
||||
packet1 := &TCPPacket{Data: []byte("GET /pub/WWW/ HTTP/1.1\nHost: www.w3.org\r\n\r\n")}
|
||||
packet2 := &TCPPacket{Data: []byte("asd=asdasd&zxc=qwe\r\n\r\n")}
|
||||
|
||||
return &TCPMessage{packets: []*TCPPacket{packet}}
|
||||
return &TCPMessage{packets: []*TCPPacket{packet1, packet2}}
|
||||
}
|
||||
|
||||
func mockReplayServer() (listener net.Listener) {
|
||||
listener, _ = net.Listen("tcp", "127.0.0.1:0")
|
||||
func mockServer() (replay net.Listener) {
|
||||
replay, _ = net.Listen("tcp", "127.0.0.1:0")
|
||||
|
||||
Settings.ReplayAddress = listener.Addr().String()
|
||||
|
||||
fmt.Println(listener.Addr().String())
|
||||
fmt.Println(replay.Addr().String())
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
func TestSendMessage(t *testing.T) {
|
||||
Settings.Verbose = true
|
||||
Settings.Verbose = false
|
||||
|
||||
listener := mockReplayServer()
|
||||
replay := mockServer()
|
||||
|
||||
Settings.ReplayAddress = replay.Addr().String()
|
||||
|
||||
msg := getTCPMessage()
|
||||
|
||||
sendMessage(msg)
|
||||
|
||||
conn, _ := listener.Accept()
|
||||
conn, _ := replay.Accept()
|
||||
defer conn.Close()
|
||||
|
||||
buf := make([]byte, 1024)
|
||||
@@ -40,6 +42,40 @@ 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")
|
||||
}
|
||||
}
|
||||
|
||||
/*
|
||||
func TestPerformance(t *testing.T) {
|
||||
Settings.Verbose = true
|
||||
|
||||
replay := mockReplayServer()
|
||||
|
||||
msg := getTCPMessage()
|
||||
|
||||
for y := 0; y < 10; y++ {
|
||||
go func() {
|
||||
for {
|
||||
conn, _ := replay.Accept()
|
||||
|
||||
go func() {
|
||||
buf := make([]byte, 1024)
|
||||
n, _ := conn.Read(buf)
|
||||
buf = buf[0:n]
|
||||
|
||||
conn.Close()
|
||||
}()
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
for y := 0; y < 10; y++ {
|
||||
for i := 0; i < 500; i++ {
|
||||
go sendMessage(msg)
|
||||
}
|
||||
|
||||
time.Sleep(time.Millisecond * 1000)
|
||||
}
|
||||
}
|
||||
*/
|
||||
|
||||
@@ -14,7 +14,7 @@ import (
|
||||
// Since we can't use default TCP libraries RAWTCPLitener implements own TCP layer
|
||||
// TCP packets is parsed using tcp_packet.go, and flow control is managed by tcp_message.go
|
||||
type RAWTCPListener struct {
|
||||
messages []*TCPMessage // buffer of TCPMessages waiting to be send
|
||||
messages map[uint32]*TCPMessage // buffer of TCPMessages waiting to be send
|
||||
|
||||
c_packets chan *TCPPacket
|
||||
c_messages chan *TCPMessage // Messages ready to be send to client
|
||||
@@ -25,12 +25,14 @@ type RAWTCPListener struct {
|
||||
port int // Port to listen
|
||||
}
|
||||
|
||||
// RAWTCPListen creates a listener to capture traffic from RAW_SOCKET
|
||||
func RAWTCPListen(addr string, port int) (rawListener *RAWTCPListener) {
|
||||
rawListener = &RAWTCPListener{}
|
||||
|
||||
rawListener.c_packets = make(chan *TCPPacket)
|
||||
rawListener.c_messages = make(chan *TCPMessage)
|
||||
rawListener.c_del_message = make(chan *TCPMessage)
|
||||
rawListener.c_packets = make(chan *TCPPacket, 100)
|
||||
rawListener.c_messages = make(chan *TCPMessage, 100)
|
||||
rawListener.c_del_message = make(chan *TCPMessage, 100)
|
||||
rawListener.messages = make(map[uint32]*TCPMessage)
|
||||
|
||||
rawListener.addr = addr
|
||||
rawListener.port = port
|
||||
@@ -47,41 +49,15 @@ func (t *RAWTCPListener) listen() {
|
||||
// If message ready for deletion it means that its also complete or expired by timeout
|
||||
case message := <-t.c_del_message:
|
||||
t.c_messages <- message
|
||||
t.deleteMessage(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 {
|
||||
var idx int = -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()
|
||||
@@ -98,62 +74,64 @@ func (t *RAWTCPListener) readRAWSocket() {
|
||||
|
||||
if err != nil {
|
||||
Debug("Error:", err)
|
||||
continue
|
||||
}
|
||||
|
||||
if n > 0 {
|
||||
// To avoid full packet parsing every time, we manually parsing values needed for packet filtering
|
||||
// http://en.wikipedia.org/wiki/Transmission_Control_Protocol
|
||||
dest_port := binary.BigEndian.Uint16(buf[2:4])
|
||||
|
||||
// Because RAW_SOCKET can't be bound to port, we have to control it by ourself
|
||||
if int(dest_port) == t.port {
|
||||
// Check TCPPacket code for more description
|
||||
flags := binary.BigEndian.Uint16(buf[12:14]) & 0x1FF
|
||||
f_psh := (flags & TCP_PSH) != 0
|
||||
|
||||
// We need only packets with data inside
|
||||
// TCP PSH flag indicate that packet have data inside
|
||||
if f_psh {
|
||||
// We should create new buffer because go slices is pointers. So buffer data shoud be immutable.
|
||||
new_buf := make([]byte, n)
|
||||
copy(new_buf, buf[:n])
|
||||
|
||||
// To avoid socket locking processing packet in new goroutine
|
||||
go func(buf []byte) {
|
||||
packet := NewTCPPacket(new_buf)
|
||||
t.c_packets <- packet
|
||||
}(new_buf)
|
||||
}
|
||||
}
|
||||
t.parsePacket(buf[:n])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (t *RAWTCPListener) parsePacket(buf []byte) {
|
||||
if t.isIncomingDataPacket(buf) {
|
||||
new_buf := make([]byte, len(buf))
|
||||
copy(new_buf, buf)
|
||||
|
||||
t.c_packets <- ParseTCPPacket(new_buf)
|
||||
}
|
||||
}
|
||||
|
||||
func (t *RAWTCPListener) isIncomingDataPacket(buf []byte) bool {
|
||||
// To avoid full packet parsing every time, we manually parsing values needed for packet filtering
|
||||
// http://en.wikipedia.org/wiki/Transmission_Control_Protocol
|
||||
dest_port := binary.BigEndian.Uint16(buf[2:4])
|
||||
|
||||
// Because RAW_SOCKET can't be bound to port, we have to control it by ourself
|
||||
if int(dest_port) == t.port {
|
||||
// Check TCPPacket code for more description
|
||||
flags := binary.BigEndian.Uint16(buf[12:14]) & 0x1FF
|
||||
|
||||
// We need only packets with data inside
|
||||
// TCP PSH flag indicate that packet have data inside
|
||||
if (flags & TCP_PSH) != 0 {
|
||||
// We should create new buffer because go slices is pointers. So buffer data shoud be immutable.
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
// Trying to add packet to existing message or creating new message
|
||||
//
|
||||
// For TCP message unique id is Acknowledgment number (see tcp_packet.go)
|
||||
func (t *RAWTCPListener) processTCPPacket(packet *TCPPacket) {
|
||||
var message *TCPMessage
|
||||
|
||||
// Searching for message with same Ack
|
||||
for _, msg := range t.messages {
|
||||
if msg.Ack == packet.Ack {
|
||||
message = msg
|
||||
break
|
||||
}
|
||||
}
|
||||
message, ok := t.messages[packet.Ack]
|
||||
|
||||
if message == nil {
|
||||
if !ok {
|
||||
// We sending c_del_message channel, so message object can communicate with Listener and notify it if message completed
|
||||
message = NewTCPMessage(packet.Ack, t.c_del_message)
|
||||
|
||||
t.messages = append(t.messages, message)
|
||||
t.messages[packet.Ack] = message
|
||||
}
|
||||
|
||||
// Adding packet to message
|
||||
message.c_packets <- packet
|
||||
}
|
||||
|
||||
// Receive TCP messages from the listener channel
|
||||
func (t *RAWTCPListener) Receive() *TCPMessage {
|
||||
return <-t.c_messages
|
||||
}
|
||||
|
||||
@@ -0,0 +1,84 @@
|
||||
package listener
|
||||
|
||||
import (
|
||||
"encoding/binary"
|
||||
"math/rand"
|
||||
"net"
|
||||
"strconv"
|
||||
"sync"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func createHeader(ack uint32, port int) (header []byte, o_ack uint32) {
|
||||
if ack == 0 {
|
||||
ack = rand.Uint32()
|
||||
}
|
||||
|
||||
seq := rand.Uint32()
|
||||
|
||||
header = make([]byte, 256)
|
||||
|
||||
binary.BigEndian.PutUint16(header[2:4], uint16(port))
|
||||
binary.BigEndian.PutUint32(header[4:8], seq)
|
||||
binary.BigEndian.PutUint32(header[8:12], ack)
|
||||
header[12] = 4 << 4
|
||||
header[13] = 8 // Setting PSH flag
|
||||
|
||||
return header, ack
|
||||
}
|
||||
|
||||
func getPackets(port int) [][]byte {
|
||||
if rand.Int()%2 == 0 {
|
||||
tcp, _ := createHeader(uint32(0), port)
|
||||
tcp = append(tcp, []byte("GET /pub/WWW/ HTTP/1.1\nHost: www.w3.org\r\n\r\n")...)
|
||||
|
||||
return [][]byte{tcp}
|
||||
} else {
|
||||
tcp1, ack := createHeader(uint32(0), port)
|
||||
tcp1 = append(tcp1, []byte("POST /pub/WWW/ HTTP/1.1\nHost: www.w3.org\r\n\r\n")...)
|
||||
|
||||
tcp2, _ := createHeader(ack, port)
|
||||
tcp2 = append(tcp2, []byte("a=1&b=2\r\n\r\n")...)
|
||||
|
||||
return [][]byte{tcp1, tcp2}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func TestRawTCPListener(t *testing.T) {
|
||||
Settings.Verbose = true
|
||||
|
||||
server := mockServer()
|
||||
//server_addr := server.Addr().String()
|
||||
host, port_str, _ := net.SplitHostPort(server.Addr().String())
|
||||
port, _ := strconv.Atoi(port_str)
|
||||
|
||||
// Accept all packets
|
||||
go func() {
|
||||
conn, _ := server.Accept()
|
||||
conn.Close()
|
||||
}()
|
||||
|
||||
listener := RAWTCPListen(host, port)
|
||||
|
||||
var wg sync.WaitGroup
|
||||
|
||||
go func() {
|
||||
for {
|
||||
listener.Receive()
|
||||
wg.Done()
|
||||
}
|
||||
}()
|
||||
|
||||
for i := 0; i < 10000; i++ {
|
||||
wg.Add(1)
|
||||
|
||||
packets := getPackets(port)
|
||||
|
||||
for _, packet := range packets {
|
||||
listener.parsePacket(packet)
|
||||
}
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
}
|
||||
@@ -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
|
||||
@@ -28,14 +29,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() {
|
||||
@@ -44,10 +46,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.StringVar(&Settings.FileToReplyPath, "file", "", "File to store captured requests")
|
||||
|
||||
|
||||
+14
-19
@@ -10,28 +10,25 @@ 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
|
||||
|
||||
timer *time.Timer // Used for expire check
|
||||
|
||||
expired bool
|
||||
|
||||
c_packets chan *TCPPacket
|
||||
c_closing chan int
|
||||
|
||||
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}
|
||||
|
||||
msg.c_packets = make(chan *TCPPacket)
|
||||
msg.c_closing = make(chan int)
|
||||
msg.c_del_message = c_del // used for notifying that message completed or expired
|
||||
|
||||
// Every time we receive packet we reset this timer
|
||||
@@ -45,21 +42,24 @@ func NewTCPMessage(Ack uint32, c_del chan *TCPMessage) (msg *TCPMessage) {
|
||||
func (t *TCPMessage) listen() {
|
||||
for {
|
||||
select {
|
||||
case <-t.c_closing:
|
||||
close(t.c_packets)
|
||||
return // Stop loop if message completed/expired
|
||||
case packet := <-t.c_packets:
|
||||
t.AddPacket(packet)
|
||||
case packet, more := <-t.c_packets:
|
||||
if more {
|
||||
t.AddPacket(packet)
|
||||
} else {
|
||||
// Stop loop if channel closed
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
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))
|
||||
|
||||
@@ -78,14 +78,9 @@ 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 {
|
||||
Debug("Adding packet to expired message")
|
||||
return
|
||||
}
|
||||
|
||||
packetFound := false
|
||||
|
||||
for _, pkt := range t.packets {
|
||||
|
||||
+9
-12
@@ -36,43 +36,40 @@ type TCPPacket struct {
|
||||
Data []byte
|
||||
}
|
||||
|
||||
func NewTCPPacket(b []byte) (p *TCPPacket) {
|
||||
func ParseTCPPacket(b []byte) (p *TCPPacket) {
|
||||
p = &TCPPacket{Data: b}
|
||||
p.ParseFast()
|
||||
p.ParseBasic()
|
||||
|
||||
return p
|
||||
}
|
||||
|
||||
// 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),
|
||||
|
||||
+33
-33
@@ -46,13 +46,14 @@ func NewReplayManager() (rm *ReplayManager) {
|
||||
return
|
||||
}
|
||||
|
||||
// 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)
|
||||
@@ -66,18 +67,17 @@ 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() {
|
||||
rm := NewReplayManager();
|
||||
rm := NewReplayManager()
|
||||
|
||||
if Settings.FileToReplyPath != "" {
|
||||
rm.RunReplayFromFile()
|
||||
} else {
|
||||
rm.RunReplayFromNetwork()
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func (self *ReplayManager) RunReplayFromFile() {
|
||||
@@ -103,9 +103,9 @@ func (self *ReplayManager) RunReplayFromFile() {
|
||||
}
|
||||
|
||||
func (self *ReplayManager) RunReplayFromNetwork() {
|
||||
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)
|
||||
@@ -129,41 +129,41 @@ func (self *ReplayManager) RunReplayFromNetwork() {
|
||||
}
|
||||
|
||||
func (self *ReplayManager) handleConnection(conn net.Conn) error {
|
||||
defer conn.Close()
|
||||
defer conn.Close()
|
||||
|
||||
var read = true
|
||||
var response []byte
|
||||
var buf []byte
|
||||
var read = true
|
||||
var response []byte
|
||||
var buf []byte
|
||||
|
||||
buf = make([]byte, bufSize)
|
||||
buf = make([]byte, bufSize)
|
||||
|
||||
for read {
|
||||
n, err := conn.Read(buf)
|
||||
for read {
|
||||
n, err := conn.Read(buf)
|
||||
|
||||
switch err {
|
||||
case io.EOF:
|
||||
read = false
|
||||
case nil:
|
||||
response = append(response, buf[0:n]...)
|
||||
if n < bufSize {
|
||||
read = false
|
||||
}
|
||||
default:
|
||||
read = false
|
||||
}
|
||||
}
|
||||
switch err {
|
||||
case io.EOF:
|
||||
read = false
|
||||
case nil:
|
||||
response = append(response, buf[:n]...)
|
||||
if n < bufSize {
|
||||
read = false
|
||||
}
|
||||
default:
|
||||
read = false
|
||||
}
|
||||
}
|
||||
|
||||
go func() {
|
||||
if request, err := ParseRequest(response); err != nil {
|
||||
Debug("Error while parsing request", err, response)
|
||||
} else {
|
||||
Debug("Adding request", request)
|
||||
go func() {
|
||||
if request, err := ParseRequest(response); err != nil {
|
||||
Debug("Error while parsing request", err, response)
|
||||
} else {
|
||||
Debug("Adding request", request)
|
||||
|
||||
self.sendRequestToReplay(request)
|
||||
}
|
||||
}()
|
||||
}
|
||||
}()
|
||||
|
||||
return nil
|
||||
return nil
|
||||
}
|
||||
|
||||
func (self *ReplayManager) sendRequestToReplay(req *http.Request) {
|
||||
|
||||
@@ -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,
|
||||
@@ -69,7 +71,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()
|
||||
|
||||
|
||||
@@ -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,7 +40,7 @@ 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 {
|
||||
@@ -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()
|
||||
|
||||
+13
-7
@@ -7,6 +7,7 @@ import (
|
||||
"strings"
|
||||
)
|
||||
|
||||
// ForwardHost where to forward requests
|
||||
type ForwardHost struct {
|
||||
Url string
|
||||
Limit int
|
||||
@@ -14,16 +15,22 @@ 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
|
||||
|
||||
FileToReplyPath 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"
|
||||
@@ -51,13 +58,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
|
||||
@@ -67,14 +72,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.StringVar(&Settings.FileToReplyPath, "file", "", "File to replay captured requests from")
|
||||
|
||||
|
||||
Reference in New Issue
Block a user