Compare commits

..
1 Commits
Author SHA1 Message Date
Leonid Bugaev c0b6d7e428 Add AF_PACKET engine 2021-06-14 20:52:26 +03:00
33 changed files with 1269 additions and 1373 deletions
+1 -3
View File
@@ -1,4 +1,3 @@
vendor
*.swp
*.gor
*.rpm
@@ -6,9 +5,8 @@ vendor
*.deb
*.pkg
*.exe
*.pprof
*.out
hey
*.bin
+20 -19
View File
@@ -23,24 +23,16 @@ FPMCOMMON= \
-s dir \
-C /tmp/gor-build \
release: release-x64 release-x86 release-mac release-windows
release: release-x64 release-mac release-windows
vendor:
go mod vendor
release-bin:
docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=amd64 -i $(CONTAINER) go build -o $(BIN_NAME) -tags netgo $(LDFLAGS)
release-bin: vendor
docker run --rm -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=amd64 -i $(CONTAINER) go build -mod=vendor -o $(BIN_NAME) -tags netgo $(LDFLAGS)
release-bin-mac:
GOOS=darwin go build -o $(BIN_NAME) $(MAC_LDFLAGS)
release-bin-x64: vendor
docker run --rm -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=386 -i $(CONTAINER) go build -mod=vendor -o $(BIN_NAME) -tags netgo $(LDFLAGS)
release-bin-mac: vendor
GOOS=darwin go build -mod=vendor -o $(BIN_NAME) $(MAC_LDFLAGS)
release-bin-windows: vendor
docker run -it --rm -v `pwd`:$(SOURCE_PATH) -w $(SOURCE_PATH) -e CGO_ENABLED=1 docker.elastic.co/beats-dev/golang-crossbuild:1.16.4-main --build-cmd "make VERSION=$(VERSION) build" -p "windows/amd64"
release-x64: release-bin
release-x64:
docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=amd64 -i $(CONTAINER) go build -o $(BIN_NAME) -tags netgo $(LDFLAGS)
tar -czf gor_$(VERSION)$(PREFIX)_x64.tar.gz $(BIN_NAME)
mkdir -p /tmp/gor-build
mv ./$(BIN_NAME) /tmp/gor-build/$(BIN_NAME)
@@ -51,11 +43,13 @@ release-x64: release-bin
fpm $(FPMCOMMON) -a amd64 -t rpm ./=/usr/local/bin
rm -rf /tmp/gor-build
release-x86: release-bin-x64
release-x86:
docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=386 -i $(CONTAINER) go build -o $(BIN_NAME) -tags netgo $(LDFLAGS)
tar -czf gor_$(VERSION)$(PREFIX)_x86.tar.gz $(BIN_NAME)
rm $(BIN_NAME)
release-mac: release-bin-mac
release-mac:
go build -o $(BIN_NAME) $(MAC_LDFLAGS)
tar -czf gor_$(VERSION)$(PREFIX)_mac.tar.gz $(BIN_NAME)
mkdir -p /tmp/gor-build
mv ./$(BIN_NAME) /tmp/gor-build/$(BIN_NAME)
@@ -64,13 +58,20 @@ release-mac: release-bin-mac
fpm $(FPMCOMMON) -a amd64 -t osxpkg ./=/usr/local/bin
rm -rf /tmp/gor-build
release-windows: release-bin-windows
release-windows:
docker run -it --rm \
-v `pwd`:/go/src/github.com/buger/goreplay \
-w /go/src/github.com/buger/goreplay \
-e CGO_ENABLED=1 \
docker.elastic.co/beats-dev/golang-crossbuild:1.16.4-main \
--build-cmd "make VERSION=$(VERSION) build" \
-p "windows/amd64"
mv ./gor ./gor.exe
zip gor-$(VERSION)$(PREFIX)_windows.zip ./gor.exe
rm -rf ./gor.exe
build:
go build -mod=vendor -o $(BIN_NAME) $(LDFLAGS)
go build -o $(BIN_NAME) $(LDFLAGS)
install:
go install $(MAC_LDFLAGS)
+163 -15
View File
@@ -1,32 +1,180 @@
// +build !linux
package capture
package main
import (
"flag"
"fmt"
"log"
"os"
"runtime/pprof"
"time"
"github.com/google/gopacket"
"github.com/google/gopacket/afpacket"
"github.com/google/gopacket/layers"
"github.com/google/gopacket/pcap"
"golang.org/x/net/bpf"
_ "github.com/google/gopacket/layers"
)
var (
iface = flag.String("i", "any", "Interface to read from")
cpuprofile = flag.String("cpuprofile", "", "If non-empty, write CPU profile here")
snaplen = flag.Int("s", 0, "Snaplen, if <= 0, use 65535")
bufferSize = flag.Int("b", 8, "Interface buffersize (MB)")
filter = flag.String("f", "port not 22", "BPF filter")
count = flag.Int64("c", -1, "If >= 0, # of packets to capture before returning")
verbose = flag.Int64("log_every", 1, "Write a log every X packets")
addVLAN = flag.Bool("add_vlan", false, "If true, add VLAN header")
)
type afpacketHandle struct {
TPacket *afpacket.TPacket
}
func newAfpacketHandle(device string, snaplen int, block_size int, num_blocks int,
useVLAN bool, timeout time.Duration) (*afpacketHandle, error) {
return nil, fmt.Errorf("Not implemented")
h := &afpacketHandle{}
var err error
if device == "any" {
h.TPacket, err = afpacket.NewTPacket(
afpacket.OptFrameSize(snaplen),
afpacket.OptBlockSize(block_size),
afpacket.OptNumBlocks(num_blocks),
afpacket.OptAddVLANHeader(useVLAN),
afpacket.OptPollTimeout(timeout),
afpacket.SocketRaw,
afpacket.TPacketVersion3)
} else {
h.TPacket, err = afpacket.NewTPacket(
afpacket.OptInterface(device),
afpacket.OptFrameSize(snaplen),
afpacket.OptBlockSize(block_size),
afpacket.OptNumBlocks(num_blocks),
afpacket.OptAddVLANHeader(useVLAN),
afpacket.OptPollTimeout(timeout),
afpacket.SocketRaw,
afpacket.TPacketVersion3)
}
return h, err
}
func afpacketComputeSize(targetSizeMb int, snaplen int, pageSize int) (
frameSize int, blockSize int, numBlocks int, err error) {
return 0, 0, 0, fmt.Errorf("Not implemented")
}
type afpacketHandle struct{}
// ReadPacketData satisfies PacketDataSource interface
func (h *afpacketHandle) ReadPacketData() (data []byte, ci gopacket.CaptureInfo, err error) {
return nil, gopacket.CaptureInfo{}, fmt.Errorf("Not implemented")
// ZeroCopyReadPacketData satisfies ZeroCopyPacketDataSource interface
func (h *afpacketHandle) ZeroCopyReadPacketData() (data []byte, ci gopacket.CaptureInfo, err error) {
return h.TPacket.ZeroCopyReadPacketData()
}
// SetBPFFilter translates a BPF filter string into BPF RawInstruction and applies them.
func (h *afpacketHandle) SetBPFFilter(filter string, snaplen int) (err error) {
return fmt.Errorf("Not implemented")
pcapBPF, err := pcap.CompileBPFFilter(layers.LinkTypeEthernet, snaplen, filter)
if err != nil {
return err
}
bpfIns := []bpf.RawInstruction{}
for _, ins := range pcapBPF {
bpfIns2 := bpf.RawInstruction{
Op: ins.Code,
Jt: ins.Jt,
Jf: ins.Jf,
K: ins.K,
}
bpfIns = append(bpfIns, bpfIns2)
}
if h.TPacket.SetBPF(bpfIns); err != nil {
return err
}
return h.TPacket.SetBPF(bpfIns)
}
// LinkType returns ethernet link type.
func (h *afpacketHandle) LinkType() layers.LinkType {
return layers.LinkTypeEthernet
}
// Close will close afpacket source.
func (h *afpacketHandle) Close() {
h.TPacket.Close()
}
// SocketStats prints received, dropped, queue-freeze packet stats.
func (h *afpacketHandle) SocketStats() (as afpacket.SocketStats, asv afpacket.SocketStatsV3, err error) {
return h.TPacket.SocketStats()
}
// afpacketComputeSize computes the block_size and the num_blocks in such a way that the
// allocated mmap buffer is close to but smaller than target_size_mb.
// The restriction is that the block_size must be divisible by both the
// frame size and page size.
func afpacketComputeSize(targetSizeMb int, snaplen int, pageSize int) (
frameSize int, blockSize int, numBlocks int, err error) {
if snaplen < pageSize {
frameSize = pageSize / (pageSize / snaplen)
} else {
frameSize = (snaplen/pageSize + 1) * pageSize
}
// 128 is the default from the gopacket library so just use that
blockSize = frameSize * 128
numBlocks = (targetSizeMb * 1024 * 1024) / blockSize
if numBlocks == 0 {
return 0, 0, 0, fmt.Errorf("Interface buffersize is too small")
}
return frameSize, blockSize, numBlocks, nil
}
func main() {
flag.Parse()
if *cpuprofile != "" {
log.Printf("Writing CPU profile to %q", *cpuprofile)
f, err := os.Create(*cpuprofile)
if err != nil {
log.Fatal(err)
}
if err := pprof.StartCPUProfile(f); err != nil {
log.Fatal(err)
}
defer pprof.StopCPUProfile()
}
log.Printf("Starting on interface %q", *iface)
if *snaplen <= 0 {
*snaplen = 65535
}
szFrame, szBlock, numBlocks, err := afpacketComputeSize(*bufferSize, *snaplen, os.Getpagesize())
if err != nil {
log.Fatal(err)
}
afpacketHandle, err := newAfpacketHandle(*iface, szFrame, szBlock, numBlocks, *addVLAN, pcap.BlockForever)
if err != nil {
log.Fatal(err)
}
err = afpacketHandle.SetBPFFilter(*filter, *snaplen)
if err != nil {
log.Fatal(err)
}
source := gopacket.ZeroCopyPacketDataSource(afpacketHandle)
defer afpacketHandle.Close()
bytes := uint64(0)
packets := uint64(0)
for ; *count != 0; *count-- {
data, _, err := source.ZeroCopyReadPacketData()
if err != nil {
log.Fatal(err)
}
bytes += uint64(len(data))
packets++
if *count%*verbose == 0 {
_, afpacketStats, err := afpacketHandle.SocketStats()
if err != nil {
log.Println(err)
}
log.Printf("Read in %d bytes in %d packets", bytes, packets)
log.Printf("Stats {received dropped queue-freeze}: %d", afpacketStats)
}
}
}
-115
View File
@@ -1,115 +0,0 @@
// +build linux
package capture
import (
"fmt"
"time"
"github.com/google/gopacket"
"github.com/google/gopacket/afpacket"
"github.com/google/gopacket/layers"
"github.com/google/gopacket/pcap"
"golang.org/x/net/bpf"
_ "github.com/google/gopacket/layers"
)
type afpacketHandle struct {
TPacket *afpacket.TPacket
}
func newAfpacketHandle(device string, snaplen int, block_size int, num_blocks int,
useVLAN bool, timeout time.Duration) (*afpacketHandle, error) {
h := &afpacketHandle{}
var err error
if device == "any" {
h.TPacket, err = afpacket.NewTPacket(
afpacket.OptFrameSize(snaplen),
afpacket.OptBlockSize(block_size),
afpacket.OptNumBlocks(num_blocks),
afpacket.OptAddVLANHeader(false),
afpacket.OptPollTimeout(timeout),
afpacket.SocketRaw,
afpacket.TPacketVersion3)
} else {
h.TPacket, err = afpacket.NewTPacket(
afpacket.OptInterface(device),
afpacket.OptFrameSize(snaplen),
afpacket.OptBlockSize(block_size),
afpacket.OptNumBlocks(num_blocks),
afpacket.OptAddVLANHeader(false),
afpacket.OptPollTimeout(timeout),
afpacket.SocketRaw,
afpacket.TPacketVersion3)
}
return h, err
}
// ReadPacketData satisfies PacketDataSource interface
func (h *afpacketHandle) ReadPacketData() (data []byte, ci gopacket.CaptureInfo, err error) {
return h.TPacket.ReadPacketData()
}
// SetBPFFilter translates a BPF filter string into BPF RawInstruction and applies them.
func (h *afpacketHandle) SetBPFFilter(filter string, snaplen int) (err error) {
pcapBPF, err := pcap.CompileBPFFilter(layers.LinkTypeEthernet, snaplen, filter)
if err != nil {
return err
}
bpfIns := []bpf.RawInstruction{}
for _, ins := range pcapBPF {
bpfIns2 := bpf.RawInstruction{
Op: ins.Code,
Jt: ins.Jt,
Jf: ins.Jf,
K: ins.K,
}
bpfIns = append(bpfIns, bpfIns2)
}
if h.TPacket.SetBPF(bpfIns); err != nil {
return err
}
return h.TPacket.SetBPF(bpfIns)
}
// LinkType returns ethernet link type.
func (h *afpacketHandle) LinkType() layers.LinkType {
return layers.LinkTypeEthernet
}
// Close will close afpacket source.
func (h *afpacketHandle) Close() {
h.TPacket.Close()
}
// SocketStats prints received, dropped, queue-freeze packet stats.
func (h *afpacketHandle) SocketStats() (as afpacket.SocketStats, asv afpacket.SocketStatsV3, err error) {
return h.TPacket.SocketStats()
}
// afpacketComputeSize computes the block_size and the num_blocks in such a way that the
// allocated mmap buffer is close to but smaller than target_size_mb.
// The restriction is that the block_size must be divisible by both the
// frame size and page size.
func afpacketComputeSize(targetSizeMb int, snaplen int, pageSize int) (
frameSize int, blockSize int, numBlocks int, err error) {
if snaplen < pageSize {
frameSize = pageSize / (pageSize / snaplen)
} else {
frameSize = (snaplen/pageSize + 1) * pageSize
}
// 128 is the default from the gopacket library so just use that
blockSize = frameSize * 128
numBlocks = (targetSizeMb * 1024 * 1024) / blockSize
if numBlocks == 0 {
return 0, 0, 0, fmt.Errorf("Interface buffersize is too small")
}
return frameSize, blockSize, numBlocks, nil
}
+94 -214
View File
@@ -3,7 +3,6 @@ package capture
import (
"context"
"errors"
"expvar"
"fmt"
"io"
"log"
@@ -15,7 +14,6 @@ import (
"syscall"
"time"
"github.com/buger/goreplay/proto"
"github.com/buger/goreplay/size"
"github.com/buger/goreplay/tcp"
@@ -24,20 +22,9 @@ import (
"github.com/google/gopacket/pcap"
)
var stats *expvar.Map
func init() {
stats = expvar.NewMap("raw")
stats.Init()
}
// PacketHandler is a function that is used to handle packets
type PacketHandler func(*tcp.Packet)
type PcapStatProvider interface {
Stats() (*pcap.Stats, error)
}
// PcapOptions options that can be set on a pcap capture handle,
// these options take effect on inactive pcap handles
type PcapOptions struct {
@@ -55,18 +42,14 @@ type Listener struct {
sync.Mutex
Transport string // transport layer default to tcp
Activate func() error // function is used to activate the engine. it must be called before reading packets
Handles map[string]packetHandle
Handles map[string]gopacket.ZeroCopyPacketDataSource
Interfaces []pcap.Interface
loopIndex int
Reading chan bool // this channel is closed when the listener has started reading packets
PcapOptions
Engine EngineType
ports []uint16 // src or/and dst ports
trackResponse bool
expiry time.Duration
allowIncomplete bool
messages chan *tcp.Message
protocol tcp.TCPProtocol
Engine EngineType
ports []uint16 // src or/and dst ports
trackResponse bool
host string // pcap file name or interface (name, hardware addr, index or ip address)
@@ -74,11 +57,6 @@ type Listener struct {
quit chan struct{}
}
type packetHandle struct {
handler gopacket.PacketDataSource
ips []net.IP
}
// EngineType ...
type EngineType uint8
@@ -126,7 +104,7 @@ func (eng *EngineType) String() (e string) {
// NewListener creates and initialize a new Listener. if transport or/and engine are invalid/unsupported
// is "tcp" and "pcap", are assumed. l.Engine and l.Transport can help to get the values used.
// if there is an error it will be associated with getting network interfaces
func NewListener(host string, ports []uint16, transport string, engine EngineType, protocol tcp.TCPProtocol, trackResponse bool, expiry time.Duration, allowIncomplete bool) (l *Listener, err error) {
func NewListener(host string, ports []uint16, transport string, engine EngineType, trackResponse bool) (l *Listener, err error) {
l = &Listener{}
l.host = host
@@ -139,16 +117,11 @@ func NewListener(host string, ports []uint16, transport string, engine EngineTyp
if transport != "" {
l.Transport = transport
}
l.Handles = make(map[string]packetHandle)
l.Handles = make(map[string]gopacket.ZeroCopyPacketDataSource)
l.trackResponse = trackResponse
l.closeDone = make(chan struct{})
l.quit = make(chan struct{})
l.Reading = make(chan bool)
l.expiry = expiry
l.allowIncomplete = allowIncomplete
l.protocol = protocol
l.messages = make(chan *tcp.Message, 10000)
switch engine {
default:
l.Engine = EnginePcap
@@ -181,8 +154,8 @@ func (l *Listener) SetPcapOptions(opts PcapOptions) {
// Listen listens for packets from the handles, and call handler on every packet received
// until the context done signal is sent or there is unrecoverable error on all handles.
// this function must be called after activating pcap handles
func (l *Listener) Listen(ctx context.Context) (err error) {
l.read()
func (l *Listener) Listen(ctx context.Context, handler PacketHandler) (err error) {
l.read(handler)
done := ctx.Done()
select {
case <-done:
@@ -195,11 +168,11 @@ func (l *Listener) Listen(ctx context.Context) (err error) {
}
// ListenBackground is like listen but can run concurrently and signal error through channel
func (l *Listener) ListenBackground(ctx context.Context) chan error {
func (l *Listener) ListenBackground(ctx context.Context, handler PacketHandler) chan error {
err := make(chan error, 1)
go func() {
defer close(err)
if e := l.Listen(ctx); err != nil {
if e := l.Listen(ctx, handler); err != nil {
err <- e
}
}()
@@ -218,7 +191,7 @@ func (l *Listener) Filter(ifi pcap.Interface) (filter string) {
filter = portsFilter(l.Transport, "dst", l.ports)
if len(hosts) != 0 && !l.Promiscuous {
if len(hosts) != 0 {
filter = fmt.Sprintf("((%s) and (%s))", filter, hostsFilter("dst", hosts))
} else {
filter = fmt.Sprintf("(%s)", filter)
@@ -227,7 +200,7 @@ func (l *Listener) Filter(ifi pcap.Interface) (filter string) {
if l.trackResponse {
responseFilter := portsFilter(l.Transport, "src", l.ports)
if len(hosts) != 0 && !l.Promiscuous {
if len(hosts) != 0 {
responseFilter = fmt.Sprintf("((%s) and (%s))", responseFilter, hostsFilter("src", hosts))
} else {
responseFilter = fmt.Sprintf("(%s)", responseFilter)
@@ -239,6 +212,23 @@ func (l *Listener) Filter(ifi pcap.Interface) (filter string) {
return
}
// PcapDumpHandler returns a handler to write packet data in PCAP
// format, See http://wiki.wireshark.org/Development/LibpcapFileFormathandler.
// if link layer is invalid Ethernet is assumed
// func PcapDumpHandler(file *os.File, link layers.LinkType) (handler func(packet *tcp.Packet) error, err error) {
// if link.String() == "" {
// link = layers.LinkTypeEthernet
// }
// w := NewWriterNanos(file)
// err = w.WriteFileHeader(64<<10, link)
// if err != nil {
// return nil, err
// }
// return func(packet *tcp.Packet) error {
// return w.WritePacket(*packet.Info, packet.Data)
// }, nil
// }
// PcapHandle returns new pcap Handle from dev on success.
// this function should be called after setting all necessary options for this listener
func (l *Listener) PcapHandle(ifi pcap.Interface) (handle *pcap.Handle, err error) {
@@ -248,11 +238,9 @@ func (l *Listener) PcapHandle(ifi pcap.Interface) (handle *pcap.Handle, err erro
return nil, fmt.Errorf("inactive handle error: %q, interface: %q", err, ifi.Name)
}
defer inactive.CleanUp()
if l.TimestampType != "" && l.TimestampType != "go" {
if l.TimestampType != "" {
var ts pcap.TimestampSource
ts, err = pcap.TimestampSourceFromString(l.TimestampType)
fmt.Println("Setting custom Timestamp Source. Supported values: `go`, ", inactive.SupportedTimestamps())
err = inactive.SetTimestampSource(ts)
if err != nil {
return nil, fmt.Errorf("%q: supported timestamps: %q, interface: %q", err, inactive.SupportedTimestamps(), ifi.Name)
@@ -295,7 +283,7 @@ func (l *Listener) PcapHandle(ifi pcap.Interface) (handle *pcap.Handle, err erro
}
}
if l.BufferTimeout == 0 {
l.BufferTimeout = 2000 * time.Millisecond
l.BufferTimeout = pcap.BlockForever
}
err = inactive.SetTimeout(l.BufferTimeout)
if err != nil {
@@ -305,16 +293,12 @@ func (l *Listener) PcapHandle(ifi pcap.Interface) (handle *pcap.Handle, err erro
if err != nil {
return nil, fmt.Errorf("PCAP Activate device error: %q, interface: %q", err, ifi.Name)
}
bpfFilter := l.BPFFilter
if bpfFilter == "" {
bpfFilter = l.Filter(ifi)
}
fmt.Println("Interface:", ifi.Name, ". BPF Filter:", bpfFilter)
err = handle.SetBPFFilter(bpfFilter)
l.BPFFilter = l.Filter(ifi)
fmt.Println("Interface:", ifi.Name, ". BPF Filter:", l.BPFFilter)
err = handle.SetBPFFilter(l.BPFFilter)
if err != nil {
handle.Close()
return nil, fmt.Errorf("BPF filter error: %q%s, interface: %q", err, bpfFilter, ifi.Name)
return nil, fmt.Errorf("BPF filter error: %q%s, interface: %q", err, l.BPFFilter, ifi.Name)
}
return
}
@@ -328,9 +312,7 @@ func (l *Listener) SocketHandle(ifi pcap.Interface) (handle Socket, err error) {
if err = handle.SetPromiscuous(l.Promiscuous || l.Monitor); err != nil {
return nil, fmt.Errorf("promiscuous mode error: %q, interface: %q", err, ifi.Name)
}
if l.BPFFilter == "" {
l.BPFFilter = l.Filter(ifi)
}
l.BPFFilter = l.Filter(ifi)
fmt.Println("BPF Filter: ", l.BPFFilter)
if err = handle.SetBPFFilter(l.BPFFilter); err != nil {
handle.Close()
@@ -340,39 +322,16 @@ func (l *Listener) SocketHandle(ifi pcap.Interface) (handle Socket, err error) {
return
}
func http1StartHint(pckt *tcp.Packet) (isRequest, isResponse bool) {
if proto.HasRequestTitle(pckt.Payload) {
return true, false
}
if proto.HasResponseTitle(pckt.Payload) {
return false, true
}
// No request or response detected
return false, false
}
func http1EndHint(m *tcp.Message) bool {
if m.MissingChunk() {
return false
}
return proto.HasFullPayload(m, m.PacketData()...)
}
func (l *Listener) read() {
func (l *Listener) read(handler PacketHandler) {
l.Lock()
defer l.Unlock()
for key, handle := range l.Handles {
go func(key string, hndl packetHandle) {
runtime.LockOSThread()
go func(key string, hndl gopacket.ZeroCopyPacketDataSource) {
defer l.closeHandles(key)
linkSize := 14
linkType := int(layers.LinkTypeEthernet)
if _, ok := hndl.handler.(*pcap.Handle); ok {
linkType = int(hndl.handler.(*pcap.Handle).LinkType())
if _, ok := hndl.(*pcap.Handle); ok {
linkType = int(hndl.(*pcap.Handle).LinkType())
linkSize, ok = pcapLinkTypeLength(linkType)
if !ok {
if os.Getenv("GORDEBUG") != "0" {
@@ -382,41 +341,17 @@ func (l *Listener) read() {
}
}
messageParser := tcp.NewMessageParser(l.messages, l.ports, hndl.ips, l.expiry, l.allowIncomplete)
if l.protocol == tcp.ProtocolHTTP {
messageParser.Start = http1StartHint
messageParser.End = http1EndHint
}
timer := time.NewTicker(1 * time.Second)
for {
select {
case <-l.quit:
return
case <-timer.C:
if h, ok := hndl.handler.(PcapStatProvider); ok {
s, err := h.Stats()
if err == nil {
stats.Add("packets_received", int64(s.PacketsReceived))
stats.Add("packets_dropped", int64(s.PacketsDropped))
stats.Add("packets_if_dropped", int64(s.PacketsIfDropped))
}
}
default:
data, ci, err := hndl.handler.ReadPacketData()
data, ci, err := hndl.ZeroCopyReadPacketData()
if err == nil {
if l.TimestampType == "go" {
ci.Timestamp = time.Now()
pckt, err := tcp.ParsePacket(data, linkType, linkSize, &ci)
if err == nil {
handler(pckt)
}
messageParser.PacketHandler(&tcp.PcapPacket{
Data: data,
LType: linkType,
LTypeLen: linkSize,
Ci: &ci,
})
continue
}
if enext, ok := err.(pcap.NextError); ok && enext == pcap.NextErrorTimeoutExpired {
@@ -442,18 +377,15 @@ func (l *Listener) read() {
close(l.Reading)
}
func (l *Listener) Messages() chan *tcp.Message {
return l.messages
}
func (l *Listener) closeHandles(key string) {
l.Lock()
defer l.Unlock()
if handle, ok := l.Handles[key]; ok {
if c, ok := handle.handler.(io.Closer); ok {
c.Close()
if _, ok = handle.(Socket); ok {
handle.(Socket).Close()
} else {
handle.(*pcap.Handle).Close()
}
delete(l.Handles, key)
if len(l.Handles) == 0 {
close(l.closeDone)
@@ -471,10 +403,37 @@ func (l *Listener) activatePcap() error {
msg += ("\n" + e.Error())
continue
}
l.Handles[ifi.Name] = packetHandle{
handler: handle,
ips: interfaceIPs(ifi),
l.Handles[ifi.Name] = handle
}
if len(l.Handles) == 0 {
return fmt.Errorf("pcap handles error:%s", msg)
}
return nil
}
func (l *Listener) activateAFPacket() error {
var e error
var msg string
szFrame, szBlock, numBlocks, err := afpacketComputeSize(*bufferSize, *snaplen, os.Getpagesize())
if l.host == "" || l.host == "any" {
handle, err := newAfpacketHandle("any", szFrame, szBlock, numBlocks, false, pcap.BlockForever)
if err != nil {
return err
}
return nil
}
for _, ifi := range l.Interfaces {
var handle *pcap.Handle
handle, err := newAfpacketHandle(ifi.Name, szFrame, szBlock, numBlocks, false, pcap.BlockForever)
if err != nil {
msg += ("\n" + err.Error())
continue
}
l.Handles[ifi.Name] = handle
}
if len(l.Handles) == 0 {
return fmt.Errorf("pcap handles error:%s", msg)
@@ -495,10 +454,7 @@ func (l *Listener) activateRawSocket() error {
msg += ("\n" + e.Error())
continue
}
l.Handles[ifi.Name] = packetHandle{
handler: handle,
ips: interfaceIPs(ifi),
}
l.Handles[ifi.Name] = handle
}
if len(l.Handles) == 0 {
return fmt.Errorf("raw socket handles error:%s", msg)
@@ -522,49 +478,10 @@ func (l *Listener) activatePcapFile() (err error) {
handle.Close()
return fmt.Errorf("BPF filter error: %q, filter: %s", e, l.BPFFilter)
}
fmt.Println("BPF Filter:", l.BPFFilter)
l.Handles["pcap_file"] = packetHandle{
handler: handle,
}
l.Handles["pcap_file"] = handle
return
}
func (l *Listener) activateAFPacket() error {
szFrame, szBlock, numBlocks, err := afpacketComputeSize(32, 32<<10, os.Getpagesize())
if err != nil {
return err
}
var msg string
for _, ifi := range l.Interfaces {
handle, err := newAfpacketHandle(ifi.Name, szFrame, szBlock, numBlocks, false, pcap.BlockForever)
if err != nil {
msg += ("\n" + err.Error())
continue
}
if l.BPFFilter == "" {
l.BPFFilter = l.Filter(ifi)
}
fmt.Println("Interface:", ifi.Name, ". BPF Filter:", l.BPFFilter)
handle.SetBPFFilter(l.BPFFilter, 64<<10)
l.Handles[ifi.Name] = packetHandle{
handler: handle,
ips: interfaceIPs(ifi),
}
}
if len(l.Handles) == 0 {
return fmt.Errorf("pcap handles error:%s", msg)
}
return nil
}
func (l *Listener) setInterfaces() (err error) {
var pifis []pcap.Interface
pifis, err = pcap.FindAllDevs()
@@ -572,67 +489,38 @@ func (l *Listener) setInterfaces() (err error) {
if err != nil {
return
}
for _, pi := range pifis {
if isDevice(l.host, pi) {
l.Interfaces = []pcap.Interface{pi}
return
}
var ni net.Interface
for _, i := range ifis {
if i.Name == pi.Name {
ni = i
break
}
addrs, _ := i.Addrs()
for _, a := range addrs {
for _, pa := range pi.Addresses {
if a.String() == pa.IP.String() {
ni = i
break
}
}
}
}
if ni.Flags&net.FlagLoopback != 0 {
if net.Flags(pi.Flags)&net.FlagLoopback != 0 {
l.loopIndex = ni.Index
}
if runtime.GOOS != "windows" {
if len(pi.Addresses) == 0 {
continue
}
if ni.Flags&net.FlagUp == 0 {
continue
if net.Flags(pi.Flags)&net.FlagUp == 0 {
continue
}
if isDevice(l.host, pi) {
l.Interfaces = []pcap.Interface{pi}
return
}
for _, addr := range pi.Addresses {
if addr.IP.String() == l.host {
l.Interfaces = []pcap.Interface{pi}
return
}
}
l.Interfaces = append(l.Interfaces, pi)
}
l.Interfaces = pifis
return
}
func isDevice(addr string, ifi pcap.Interface) bool {
// Windows npcap loopback have no IPs
if addr == "127.0.0.1" && ifi.Name == `\Device\NPF_Loopback` {
return true
}
if addr == ifi.Name {
return true
}
for _, _addr := range ifi.Addresses {
if _addr.IP.String() == addr {
return true
}
}
return false
return addr == ifi.Name
}
func interfaceAddresses(ifi pcap.Interface) []string {
@@ -643,14 +531,6 @@ func interfaceAddresses(ifi pcap.Interface) []string {
return hosts
}
func interfaceIPs(ifi pcap.Interface) []net.IP {
var ips []net.IP
for _, addr := range ifi.Addresses {
ips = append(ips, addr.IP)
}
return ips
}
func listenAll(addr string) bool {
switch addr {
case "", "0.0.0.0", "[::]", "::":
+313
View File
@@ -0,0 +1,313 @@
package capture
import (
"context"
"encoding/binary"
"io/ioutil"
"net"
"os"
"sync/atomic"
"testing"
"time"
"github.com/google/gopacket/layers"
"github.com/google/gopacket/pcap"
)
var LoopBack = func() net.Interface {
ifis, _ := net.Interfaces()
for _, v := range ifis {
if v.Flags&net.FlagLoopback != 0 {
return v
}
}
return ifis[0]
}()
func TestSetInterfaces(t *testing.T) {
l := &Listener{}
l.host = "127.0.0.1"
l.setInterfaces()
if len(l.Interfaces) != 1 {
t.Error("expected a single interface")
}
l.host = LoopBack.HardwareAddr.String()
l.setInterfaces()
if l.Interfaces[0].Name != LoopBack.Name && len(l.Interfaces) != 1 {
t.Error("interface should be loop back interface")
}
l.host = ""
l.setInterfaces()
if len(l.Interfaces) < 1 {
t.Error("should get all interfaces")
}
}
func TestBPFFilter(t *testing.T) {
l := &Listener{}
l.host = "127.0.0.1"
l.Transport = "tcp"
l.setInterfaces()
filter := l.Filter(l.Interfaces[0])
if filter != "(tcp dst portrange 0-65535 and host 127.0.0.1)" {
t.Error("wrong filter", filter)
}
l.port = 8000
l.trackResponse = true
filter = l.Filter(l.Interfaces[0])
if filter != "(tcp port 8000 and host 127.0.0.1)" {
t.Error("wrong filter")
}
}
func TestPcapDump(t *testing.T) {
f, err := ioutil.TempFile("", "pcap_file")
if err != nil {
t.Error(err)
}
h, _ := PcapDumpHandler(f, layers.LinkTypeLoop)
packets := Packets(1, 5, 5, 4)
for i := 0; i < len(packets); i++ {
if i == 1 {
// change dst port
binary.BigEndian.PutUint16(packets[i].TransLayer[2:], 8001)
}
if i == 4 {
packets[i].Info.CaptureLength = 40
}
h(packets[i])
}
name := f.Name()
f.Close()
testPcapDumpEngine(name, t)
}
func testPcapDumpEngine(f string, t *testing.T) {
defer os.Remove(f)
l, err := NewListener(f, 8000, "", EnginePcapFile, true)
err = l.Activate()
if err != nil {
t.Errorf("expected error to be nil, got %q", err)
return
}
pckts := 0
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
err = l.Listen(ctx, func(packet *Packet) {
if packet.Info.CaptureLength != 57 {
t.Errorf("expected packet length to be %d, got %d", 57, packet.Info.CaptureLength)
}
pckts++
})
if err != nil {
t.Errorf("expected error to be nil, got %q", err)
}
if pckts != 3 {
t.Errorf("expected %d packets, got %d packets", 3, pckts)
}
}
func TestPcapHandler(t *testing.T) {
l, err := NewListener(LoopBack.Name, 8000, "", EnginePcap, true)
if err != nil {
t.Errorf("expected error to be nil, got %v", err)
return
}
err = l.Activate()
if err != nil {
t.Errorf("expected error to be nil, got %v", err)
return
}
defer l.Handles[LoopBack.Name].(*pcap.Handle).Close()
if err != nil {
t.Errorf("expected error to be nil, got %v", err)
return
}
for i := 0; i < 5; i++ {
_, _ = net.Dial("tcp", "127.0.0.1:8000")
}
sts, _ := l.Handles[LoopBack.Name].(*pcap.Handle).Stats()
if sts.PacketsReceived < 5 {
t.Errorf("expected >=5 packets got %d", sts.PacketsReceived)
}
}
// func TestSocketHandler(t *testing.T) {
// l, err := NewListener(LoopBack.Name, 8000, "", EngineRawSocket, true)
// err = l.Activate()
// if err != nil {
// return
// }
// defer l.Handles[LoopBack.Name].(*SockRaw).Close()
// if err != nil {
// t.Errorf("expected error to be nil, got %v", err)
// return
// }
// for i := 0; i < 5; i++ {
// _, _ = net.Dial("tcp", "127.0.0.1:8000")
// }
// sts, _ := l.Handles[LoopBack.Name].(*SockRaw).Stats()
// if sts.Packets < 5 {
// t.Errorf("expected >=5 packets got %d", sts.Packets)
// }
// }
func BenchmarkPcapDump(b *testing.B) {
f, err := ioutil.TempFile("", "pcap_file")
if err != nil {
b.Error(err)
return
}
now := time.Now()
defer os.Remove(f.Name())
h, _ := PcapDumpHandler(f, layers.LinkTypeLoop)
packets := Packets(1, b.N, 5, 4)
for i := 0; i < len(packets); i++ {
h(packets[i])
}
f.Close()
b.Logf("%d packets in %s", b.N, time.Since(now))
}
func BenchmarkPcapFile(b *testing.B) {
f, err := ioutil.TempFile("", "pcap_file")
if err != nil {
b.Error(err)
return
}
defer os.Remove(f.Name())
h, _ := PcapDumpHandler(f, layers.LinkTypeLoop)
packets := Packets(1, b.N, 5, 4)
for i := 0; i < len(packets); i++ {
h(packets[i])
}
name := f.Name()
f.Close()
b.ResetTimer()
var l *Listener
l, err = NewListener(name, 8000, "", EnginePcapFile, true)
if err != nil {
b.Error(err)
return
}
err = l.Activate()
if err != nil {
b.Error(err)
return
}
now := time.Now()
pckts := 0
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
if err = l.Listen(ctx, func(packet *Packet) {
if packet.Info.CaptureLength != 49 {
b.Errorf("expected packet length to be %d, got %d", 49, packet.Info.CaptureLength)
}
pckts++
}); err != nil {
b.Error(err)
}
b.Logf("%d/%d packets in %s", pckts, b.N, time.Since(now))
}
// used to benchmark sock engine
var buf [1024]byte
func init() {
for i := 0; i < len(buf); i++ {
buf[i] = 0xff
}
}
func handler(n, counter *int32) PacketHandler {
return func(p *Packet) {
nn := int32(len(p.Data))
atomic.AddInt32(n, nn)
atomic.AddInt32(counter, 1)
}
}
func BenchmarkPcap(b *testing.B) {
var err error
n := new(int32)
counter := new(int32)
l, err := NewListener(LoopBack.Name, 8000, "", EnginePcap, false)
if err != nil {
b.Error(err)
return
}
l.PcapOptions.BPFFilter = "udp dst port 8000 and host 127.0.0.1"
err = l.Activate()
if err != nil {
b.Error(err)
return
}
errCh := l.ListenBackground(context.Background(), handler(n, counter))
select {
case <-l.Reading:
case err = <-errCh:
b.Error(err)
return
}
var conn net.Conn
b.ResetTimer()
for i := 0; i < b.N; i++ {
b.StopTimer()
conn, err = net.Dial("udp", "127.0.0.1:8000")
if err != nil {
b.Error(err)
return
}
b.StartTimer()
_, err = conn.Write(buf[:])
if err != nil {
b.Error(err)
return
}
}
b.ReportMetric(float64(atomic.LoadInt32(n)), "buf")
b.ReportMetric(float64(atomic.LoadInt32(counter)), "packets")
}
func BenchmarkRawSocket(b *testing.B) {
var err error
n := new(int32)
counter := new(int32)
l, err := NewListener(LoopBack.Name, 8000, "", EngineRawSocket, false)
if err != nil {
b.Error(err)
return
}
l.PcapOptions.BPFFilter = "udp dst port 8000 and host 127.0.0.1"
err = l.Activate()
if err != nil {
b.Error(err)
return
}
errCh := l.ListenBackground(context.Background(), handler(n, counter))
select {
case <-l.Reading:
case err = <-errCh:
b.Error(err)
return
}
var conn net.Conn
b.ResetTimer()
for i := 0; i < b.N; i++ {
b.StopTimer()
conn, err = net.Dial("udp", "127.0.0.1:8000")
if err != nil {
b.Error(err)
return
}
b.StartTimer()
_, err = conn.Write(buf[:])
if err != nil {
b.Error(err)
return
}
}
b.ReportMetric(float64(atomic.LoadInt32(n)), "buf")
b.ReportMetric(float64(atomic.LoadInt32(counter)), "packets")
}
+197
View File
@@ -0,0 +1,197 @@
package capture
import (
"encoding/binary"
"errors"
"testing"
"time"
"github.com/google/gopacket"
"github.com/google/gopacket/layers"
)
func generateHeader4(seq uint32, length uint16) []byte {
hdr := make([]byte, 4+24+24, 4+24+24)
binary.BigEndian.PutUint32(hdr, uint32(layers.ProtocolFamilyIPv4))
ip := hdr[4:]
ip[0] = 4<<4 | 6
binary.BigEndian.PutUint16(ip[2:4], length+24+24)
ip[9] = uint8(layers.IPProtocolTCP)
copy(ip[12:16], []byte{127, 0, 0, 1})
copy(ip[16:], []byte{127, 0, 0, 1})
// set tcp header
tcp := ip[24:]
tcp[12] = 6 << 4
binary.BigEndian.PutUint16(tcp, 5535)
binary.BigEndian.PutUint16(tcp[2:], 8000)
binary.BigEndian.PutUint32(tcp[4:], seq)
return hdr
}
func generateHeader6(seq uint32, length uint16) []byte {
hdr := make([]byte, 4+40+32+24, 4+40+32+24)
binary.BigEndian.PutUint32(hdr, uint32(layers.ProtocolFamilyIPv6Linux))
ip := hdr[4:]
ip[0] = 6 << 4
binary.BigEndian.PutUint16(ip[4:], length+32+24)
var ipAddr [16]byte
ipAddr[15] = 0x01
copy(ip[8:], ip[:])
copy(ip[24:], ip[:])
copy(ip[40:], []byte{
// net-layer (IPv6-Opts)
0x2b, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
uint8(layers.IPProtocolTCP), 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
})
// set tcp header
tcp := ip[40+32:]
tcp[12] = 6 << 4
binary.BigEndian.PutUint16(tcp, 5535)
binary.BigEndian.PutUint16(tcp[2:], 8000)
binary.BigEndian.PutUint32(tcp[4:], seq)
return hdr
}
func Packets(start uint32, _len int, length uint16, version byte) []*Packet {
var packets = make([]*Packet, _len)
for i := start; i < start+uint32(_len); i++ {
var h []byte
if version == 4 {
h = generateHeader4(i, length)
} else {
h = generateHeader6(i, length)
}
d := append(h, make([]byte, int(length))...)
ci := &gopacket.CaptureInfo{Length: len(d), CaptureLength: len(d), Timestamp: time.Now()}
packets[i-start] = NewPacket(d, int(layers.LinkTypeLoop), 4, ci)
}
return packets
}
func TestIPv4Packet(t *testing.T) {
pckt := packet(append(generateHeader4(1024, 10), make([]byte, 10)...))
if pckt.Err != nil {
t.Error(pckt)
return
}
if err := packet(pckt.Data[:2]).Err; !errors.Is(err, ErrHdrLength("Link")) {
t.Errorf("should fail with %q, got %q", ErrHdrLength("Link"), err)
return
}
if err := packet(pckt.Data[:20]).Err; !errors.Is(err, ErrHdrLength("IPv4")) {
t.Errorf("should fail with %q, got %q", ErrHdrLength("IPv4"), err)
return
}
if err := packet(pckt.Data[:27]).Err; !errors.Is(err, ErrHdrLength("IPv4 opts")) {
t.Errorf("should fail with %q, got %q", ErrHdrLength("IPv4 opts"), err)
return
}
if err := packet(pckt.Data[:40]).Err; !errors.Is(err, ErrHdrLength("TCP")) {
t.Errorf("should fail with %q, got %q", ErrHdrLength("TCP opts"), err)
return
}
if err := packet(pckt.Data[:50]).Err; !errors.Is(err, ErrHdrLength("TCP opts")) {
t.Errorf("should fail with %q, got %q", ErrHdrLength("TCP opts"), err)
return
}
pckt.TransLayer[12] = 0x10
if err := packet(pckt.Data[:50]).Err; !errors.Is(err, ErrHdrInvalid("TCP's data offset")) {
t.Errorf("should fail with %q, got %q", ErrHdrInvalid("TCP's data offset"), err)
return
}
pckt.TransLayer[12] = 0x60
if err := packet(pckt.Data[:28]).Err; !errors.Is(err, ErrHdrMissing("TCP")) {
t.Errorf("should fail with %q, got %q", ErrHdrMissing("TCP"), err)
return
}
pckt.NetLayer[9] = 0x02
if err := packet(pckt.Data).Err; !errors.Is(err, ErrHdrExpected("TCP")) {
t.Errorf("should fail with %q, got %q", ErrHdrExpected("TCP"), err)
return
}
pckt.NetLayer[9] = 0x06
pckt.NetLayer[0] = 0x44
if err := packet(pckt.Data).Err; !errors.Is(err, ErrHdrInvalid("IPv4's IHL")) {
t.Errorf("should fail with %q, got %q", ErrHdrInvalid("IPv4's IHL"), err)
return
}
pckt.NetLayer[0] = 0x56
if err := packet(pckt.Data).Err; !errors.Is(err, ErrHdrExpected("IPv4 or IPv6")) {
t.Errorf("should fail with %q, got %q", ErrHdrExpected("IPv4 or IPv6"), err)
return
}
}
func TestIPv6Packet(t *testing.T) {
pckt := packet(append(generateHeader6(1024, 10), make([]byte, 10)...))
if pckt.Err != nil {
t.Error(pckt)
return
}
if err := packet(pckt.Data[:4]).Err; !errors.Is(err, ErrHdrMissing("IPv4 or IPv6")) {
t.Errorf("should fail with %q, got %q", ErrHdrMissing("IPv4 or IPv6"), err)
return
}
if err := packet(pckt.Data[:40]).Err; !errors.Is(err, ErrHdrLength("IPv6")) {
t.Errorf("should fail with %q, got %q", ErrHdrLength("IPv6"), err)
return
}
if err := packet(pckt.Data[:52]).Err; !errors.Is(err, ErrHdrLength("IPv6 opts")) {
t.Errorf("should fail with %q, got %q", ErrHdrLength("IPv6 opts"), err)
return
}
if err := packet(pckt.Data[:80]).Err; !errors.Is(err, ErrHdrLength("TCP")) {
t.Errorf("should fail with %q, got %q", ErrHdrLength("TCP opts"), err)
return
}
if err := packet(pckt.Data[:98]).Err; !errors.Is(err, ErrHdrLength("TCP opts")) {
t.Errorf("should fail with %q, got %q", ErrHdrLength("TCP opts"), err)
return
}
pckt.TransLayer[12] = 0x10
if err := packet(pckt.Data).Err; !errors.Is(err, ErrHdrInvalid("TCP's data offset")) {
t.Errorf("should fail with %q, got %q", ErrHdrInvalid("TCP's data offset"), err)
return
}
pckt.TransLayer[12] = 0x60
if err := packet(pckt.Data[:76]).Err; !errors.Is(err, ErrHdrMissing("TCP")) {
t.Errorf("should fail with %q, got %q", ErrHdrMissing("TCP"), err)
return
}
pckt.NetLayer[56] = 0x02
if err := packet(pckt.Data).Err; !errors.Is(err, ErrHdrExpected("TCP")) {
t.Errorf("should fail with %q, got %q", ErrHdrExpected("TCP"), err)
return
}
pckt.NetLayer[56] = 0x06
}
func packet(data []byte) *Packet {
return NewPacket(data, int(layers.LinkTypeLoop), 4, &gopacket.CaptureInfo{})
}
func BenchmarkNewPacketIPv4(b *testing.B) {
data := append(generateHeader4(1204, 10), make([]byte, 10)...)
b.ResetTimer()
for i := 0; i < b.N; i++ {
if err := packet(data).Err; err != nil {
b.Error(err)
}
}
}
func BenchmarkNewPacketIPv6(b *testing.B) {
data := append(generateHeader6(1024, 10), make([]byte, 10)...)
b.ResetTimer()
for i := 0; i < b.N; i++ {
if err := packet(data).Err; err != nil {
b.Error(err)
}
}
}
+2 -4
View File
@@ -1,5 +1,3 @@
// +build linux
package capture
import (
@@ -60,7 +58,7 @@ func NewSocket(pifi pcap.Interface) (*SockRaw, error) {
}
if !found {
return nil, fmt.Errorf("can't find matching interface")
return nil, fmt.Errorf("Can't find matching interface")
}
// sock create
@@ -126,7 +124,7 @@ func NewSocket(pifi pcap.Interface) (*SockRaw, error) {
}
// ReadPacketData implements gopacket.PacketDataSource.
func (sock *SockRaw) ReadPacketData() (buf []byte, ci gopacket.CaptureInfo, err error) {
func (sock *SockRaw) ZeroCopyReadPacketData() (buf []byte, ci gopacket.CaptureInfo, err error) {
sock.mu.Lock()
defer sock.mu.Unlock()
var tpHdr *unix.Tpacket2Hdr
+1 -1
View File
@@ -8,7 +8,7 @@ import (
// Socket is any interface that defines the behaviors of Socket
type Socket interface {
ReadPacketData() ([]byte, gopacket.CaptureInfo, error)
ZeroCopyReadPacketData() ([]byte, gopacket.CaptureInfo, error)
WritePacketData([]byte) error
SetBPFFilter(string) error
SetPromiscuous(bool) error
+1 -1
View File
@@ -152,7 +152,7 @@ func CopyMulty(src PluginReader, writers ...PluginWriter) error {
}
} else {
for _, dst := range writers {
if _, err := dst.PluginWrite(msg); err != nil && err != io.ErrClosedPipe {
if _, err := dst.PluginWrite(msg); err != nil {
return err
}
}
+2 -2
View File
@@ -8,13 +8,13 @@ require (
github.com/aws/aws-sdk-go v1.33.2
github.com/bitly/go-hostpool v0.1.0 // indirect
github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 // indirect
github.com/google/gopacket v1.1.20-0.20210429153827-3eaba0894325
github.com/google/gopacket v1.1.18
github.com/klauspost/compress v1.10.10 // indirect
github.com/mattbaird/elastigo v0.0.0-20170123220020-2fe47fd29e4b
github.com/pierrec/lz4 v2.5.2+incompatible // indirect
github.com/rcrowley/go-metrics v0.0.0-20200313005456-10cdbea86bc0 // indirect
github.com/smartystreets/goconvey v1.6.4 // indirect
github.com/stretchr/testify v1.5.1
golang.org/x/net v0.0.0-20200707034311-ab3426394381
golang.org/x/net v0.0.0-20200707034311-ab3426394381 // indirect
golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd
)
+3 -14
View File
@@ -29,8 +29,8 @@ github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEW
github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
github.com/google/gopacket v1.1.20-0.20210429153827-3eaba0894325 h1:YmIcZ5Var3BAQ64AW98Iiys5Ih4fiU0xK41+8isC5Ec=
github.com/google/gopacket v1.1.20-0.20210429153827-3eaba0894325/go.mod h1:riddUzxTSBpJXk3qBHtYr4qOhFhT6k/1c0E3qkQjQpA=
github.com/google/gopacket v1.1.18 h1:lum7VRA9kdlvBi7/v2p7/zcbkduHaCH/SVVyurs7OpY=
github.com/google/gopacket v1.1.18/go.mod h1:UdDNZ1OO62aGYVnPhxT1U6aI7ukYtA/kB8vaU0diBUM=
github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1 h1:EGx4pi6eqNxGaHF6qqu48+N2wcFQ5qg5FXgOdqsJ5d8=
github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
github.com/hashicorp/go-uuid v1.0.2 h1:cfejS+Tpcp13yd5nYHWDI6qVCny6wyX2Mt5SGur2IGE=
@@ -69,29 +69,20 @@ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
github.com/stretchr/testify v1.5.1 h1:nOGnQDM7FYENwehXlg/kFVnos3rEvtKTjRvOWSzb6H4=
github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
github.com/vishvananda/netlink v1.1.0/go.mod h1:cTgwzPIzzgDAYoQrMm0EdrjRUBkTqKYppBueQtXaqoE=
github.com/vishvananda/netns v0.0.0-20191106174202-0a2b9b5464df/go.mod h1:JP3t17pCcGlemwknint6hfoeCVQrEMVwxRLRjXpq+BU=
github.com/vishvananda/netns v0.0.0-20210104183010-2eb08e3e575f/go.mod h1:DD4vA1DwXk04H54A1oHXtwZmA0grkVMdPxx/VGLCah0=
github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c/go.mod h1:lB8K/P019DLNhemzwFU4jHLhdvlE6uDZjXFejJXr49I=
github.com/xdg/stringprep v1.0.0/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0HrGL1Y=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
golang.org/x/crypto v0.0.0-20200204104054-c9f3fb736b72/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9 h1:psW17arqaxU48Z5kZ0CQnkZWQJsqcURM6tKiBApRjXI=
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
golang.org/x/lint v0.0.0-20200302205851-738671d3881b/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY=
golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20200202094626-16171245cfb2/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20200707034311-ab3426394381 h1:VXak5I6aEWmAXeQjA+QSZzlgNrpq9mjcfDemuexIKsU=
golang.org/x/net v0.0.0-20200707034311-ab3426394381/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190405154228-4b34438f7a67/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20190606203320-7fc4e5ec1444/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20200217220822-9197077df867/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd h1:xhmwyvizuTgC2qz7ZlMluP20uW+C3Rm0FD/WLDX8884=
golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -99,8 +90,6 @@ golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs=
golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
golang.org/x/tools v0.0.0-20200130002326-2f3ba24bd6e7/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+2 -32
View File
@@ -3,16 +3,15 @@
package main
import (
"expvar"
"flag"
"fmt"
"log"
"net/http"
"net/http/httputil"
httppptof "net/http/pprof"
_ "net/http/pprof"
"os"
"os/signal"
"runtime"
_ "runtime/debug"
"runtime/pprof"
"syscall"
"time"
@@ -23,35 +22,6 @@ var (
memprofile = flag.String("memprofile", "", "write memory profile to this file")
)
func init() {
var defaultServeMux http.ServeMux
http.DefaultServeMux = &defaultServeMux
http.HandleFunc("/debug/vars", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json; charset=utf-8")
fmt.Fprintf(w, "{\n")
first := true
expvar.Do(func(kv expvar.KeyValue) {
if kv.Key == "memstats" || kv.Key == "cmdline" {
return
}
if !first {
fmt.Fprintf(w, ",\n")
}
first = false
fmt.Fprintf(w, "%q: %s", kv.Key, kv.Value)
})
fmt.Fprintf(w, "\n}\n")
})
http.HandleFunc("/debug/pprof/", httppptof.Index)
http.HandleFunc("/debug/pprof/cmdline", httppptof.Cmdline)
http.HandleFunc("/debug/pprof/profile", httppptof.Profile)
http.HandleFunc("/debug/pprof/symbol", httppptof.Symbol)
http.HandleFunc("/debug/pprof/trace", httppptof.Trace)
}
func loggingMiddleware(addr string, next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == "/loop" {
+25 -189
View File
@@ -4,12 +4,9 @@ import (
"bufio"
"bytes"
"compress/gzip"
"container/heap"
"errors"
"expvar"
"fmt"
"io"
"math"
"os"
"path/filepath"
"strconv"
@@ -23,73 +20,27 @@ import (
"github.com/aws/aws-sdk-go/service/s3"
)
type filePayload struct {
data []byte
timestamp int64
}
// An IntHeap is a min-heap of ints.
type payloadQueue struct {
sync.RWMutex
s []*filePayload
}
func (h payloadQueue) Len() int { return len(h.s) }
func (h payloadQueue) Less(i, j int) bool { return h.s[i].timestamp < h.s[j].timestamp }
func (h payloadQueue) Swap(i, j int) { h.s[i], h.s[j] = h.s[j], h.s[i] }
func (h *payloadQueue) Push(x interface{}) {
// Push and Pop use pointer receivers because they modify the slice's length,
// not just its contents.
h.s = append(h.s, x.(*filePayload))
}
func (h *payloadQueue) Pop() interface{} {
old := h.s
n := len(old)
x := old[n-1]
h.s = old[0 : n-1]
return x
}
func (h payloadQueue) Idx(i int) *filePayload {
return h.s[i]
}
type fileInputReader struct {
reader *bufio.Reader
data []byte
file io.ReadCloser
timestamp int64
closed int32 // Value of 0 indicates that the file is still open.
s3 bool
queue payloadQueue
readDepth int
dryRun bool
path string
}
func (f *fileInputReader) parse(init chan struct{}) error {
func (f *fileInputReader) parseNext() error {
payloadSeparatorAsBytes := []byte(payloadSeparator)
var buffer bytes.Buffer
var initialized bool
lineNum := 0
for {
line, err := f.reader.ReadBytes('\n')
lineNum++
if err != nil {
if err != io.EOF {
Debug(1, err)
} else {
f.Close()
}
f.Close()
if !initialized {
close(init)
initialized = true
}
return err
}
@@ -97,61 +48,21 @@ func (f *fileInputReader) parse(init chan struct{}) error {
asBytes := buffer.Bytes()
meta := payloadMeta(asBytes)
if len(meta) < 3 {
Debug(1, fmt.Sprintf("Found malformed record, file: %s, line %d", f.path, lineNum))
buffer = bytes.Buffer{}
continue
}
f.timestamp, _ = strconv.ParseInt(string(meta[2]), 10, 64)
f.data = asBytes[:len(asBytes)-1]
timestamp, _ := strconv.ParseInt(string(meta[2]), 10, 64)
data := asBytes[:len(asBytes)-1]
f.queue.Lock()
heap.Push(&f.queue, &filePayload{
timestamp: timestamp,
data: data,
})
f.queue.Unlock()
for {
if f.queue.Len() < f.readDepth {
break
}
if !initialized {
close(init)
initialized = true
}
if !f.dryRun {
time.Sleep(100 * time.Millisecond)
}
}
buffer = bytes.Buffer{}
continue
return nil
}
buffer.Write(line)
}
}
func (f *fileInputReader) wait() {
for {
if atomic.LoadInt32(&f.closed) == 1 {
return
}
func (f *fileInputReader) ReadPayload() []byte {
defer f.parseNext()
if f.queue.Len() > 0 {
return
}
if !f.dryRun {
time.Sleep(100 * time.Millisecond)
}
}
return
return f.data
}
// Close closes this plugin
@@ -164,7 +75,7 @@ func (f *fileInputReader) Close() error {
return nil
}
func newFileInputReader(path string, readDepth int, dryRun bool) *fileInputReader {
func newFileInputReader(path string) *fileInputReader {
var file io.ReadCloser
var err error
@@ -179,7 +90,7 @@ func newFileInputReader(path string, readDepth int, dryRun bool) *fileInputReade
return nil
}
r := &fileInputReader{path: path, file: file, closed: 0, readDepth: readDepth, dryRun: dryRun}
r := &fileInputReader{file: file, closed: 0}
if strings.HasSuffix(path, ".gz") {
gzReader, err := gzip.NewReader(file)
if err != nil {
@@ -191,11 +102,7 @@ func newFileInputReader(path string, readDepth int, dryRun bool) *fileInputReade
r.reader = bufio.NewReader(file)
}
heap.Init(&r.queue)
init := make(chan struct{})
go r.parse(init)
<-init
r.parseNext()
return r
}
@@ -209,25 +116,16 @@ type FileInput struct {
readers []*fileInputReader
speedFactor float64
loop bool
readDepth int
dryRun bool
maxWait time.Duration
stats *expvar.Map
}
// NewFileInput constructor for FileInput. Accepts file path as argument.
func NewFileInput(path string, loop bool, readDepth int, maxWait time.Duration, dryRun bool) (i *FileInput) {
func NewFileInput(path string, loop bool) (i *FileInput) {
i = new(FileInput)
i.data = make(chan []byte, 1000)
i.exit = make(chan bool)
i.path = path
i.speedFactor = 1
i.loop = loop
i.readDepth = readDepth
i.stats = expvar.NewMap("file-" + path)
i.dryRun = dryRun
i.maxWait = maxWait
if err := i.init(); err != nil {
return
@@ -267,6 +165,7 @@ func (i *FileInput) init() (err error) {
} else if matches, err = filepath.Glob(i.path); err != nil {
Debug(0, "[INPUT-FILE] Wrong file pattern", i.path, err)
return
}
if len(matches) == 0 {
@@ -277,11 +176,9 @@ func (i *FileInput) init() (err error) {
i.readers = make([]*fileInputReader, len(matches))
for idx, p := range matches {
i.readers[idx] = newFileInputReader(p, i.readDepth, i.dryRun)
i.readers[idx] = newFileInputReader(p)
}
i.stats.Add("reader_count", int64(len(matches)))
return nil
}
@@ -292,7 +189,6 @@ func (i *FileInput) PluginRead() (*Message, error) {
case <-i.exit:
return nil, ErrorStopped
case buf := <-i.data:
i.stats.Add("read_from", 1)
msg.Meta, msg.Data = payloadMetaWithBody(buf)
return &msg, nil
}
@@ -305,17 +201,11 @@ func (i *FileInput) String() string {
// Find reader with smallest timestamp e.g next payload in row
func (i *FileInput) nextReader() (next *fileInputReader) {
for _, r := range i.readers {
if r == nil {
if r == nil || atomic.LoadInt32(&r.closed) != 0 {
continue
}
r.wait()
if r.queue.Len() == 0 {
continue
}
if next == nil || r.queue.Idx(0).timestamp < next.queue.Idx(0).timestamp {
if next == nil || r.timestamp < next.timestamp {
next = r
continue
}
@@ -327,11 +217,6 @@ func (i *FileInput) nextReader() (next *fileInputReader) {
func (i *FileInput) emit() {
var lastTime int64 = -1
var maxWait, firstWait, minWait int64
minWait = math.MaxInt64
i.stats.Add("negative_wait", 0)
for {
select {
case <-i.exit:
@@ -351,48 +236,17 @@ func (i *FileInput) emit() {
}
}
reader.queue.RLock()
payload := heap.Pop(&reader.queue).(*filePayload)
i.stats.Add("total_counter", 1)
i.stats.Add("total_bytes", int64(len(payload.data)))
reader.queue.RUnlock()
if lastTime != -1 {
diff := payload.timestamp - lastTime
if firstWait == 0 {
firstWait = diff
}
diff := reader.timestamp - lastTime
lastTime = reader.timestamp
if i.speedFactor != 1 {
diff = int64(float64(diff) / i.speedFactor)
}
if i.maxWait > 0 && diff > int64(i.maxWait) {
diff = int64(i.maxWait)
}
if diff >= 0 {
lastTime = payload.timestamp
if !i.dryRun {
time.Sleep(time.Duration(diff))
}
i.stats.Add("total_wait", diff)
if diff > maxWait {
maxWait = diff
}
if diff < minWait {
minWait = diff
}
} else {
i.stats.Add("negative_wait", 1)
}
time.Sleep(time.Duration(diff))
} else {
lastTime = payload.timestamp
lastTime = reader.timestamp
}
// Recheck if we have exited since last check.
@@ -400,30 +254,12 @@ func (i *FileInput) emit() {
case <-i.exit:
return
default:
if !i.dryRun {
i.data <- payload.data
}
i.data <- reader.ReadPayload()
}
}
i.stats.Set("first_wait", time.Duration(firstWait))
i.stats.Set("max_wait", time.Duration(maxWait))
i.stats.Set("min_wait", time.Duration(minWait))
Debug(0, fmt.Sprintf("[INPUT-FILE] FileInput: end of file '%s'\n", i.path))
if i.dryRun {
fmt.Printf("Records found: %v\nFiles processed: %v\nBytes processed: %v\nMax wait: %v\nMin wait: %v\nFirst wait: %v\nIt will take `%v` to replay at current speed.\nFound %v records with out of order timestamp\n",
i.stats.Get("total_counter"),
i.stats.Get("reader_count"),
i.stats.Get("total_bytes"),
i.stats.Get("max_wait"),
i.stats.Get("min_wait"),
i.stats.Get("first_wait"),
time.Duration(i.stats.Get("total_wait").(*expvar.Int).Value()),
i.stats.Get("negative_wait"),
)
}
}
// Close closes this plugin
+6 -6
View File
@@ -104,7 +104,7 @@ func TestInputFileMultipleFilesWithRequestsOnly(t *testing.T) {
file2.Write([]byte(payloadSeparator))
file2.Close()
input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100, 0, false)
input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false)
for i := '1'; i <= '4'; i++ {
msg, _ := input.PluginRead()
@@ -130,7 +130,7 @@ func TestInputFileRequestsWithLatency(t *testing.T) {
file.Write([]byte("1 3 250000000\nrequest3"))
file.Write([]byte(payloadSeparator))
input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), false, 100, 0, false)
input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), false)
start := time.Now().UnixNano()
for i := 0; i < 3; i++ {
@@ -170,7 +170,7 @@ func TestInputFileMultipleFilesWithRequestsAndResponses(t *testing.T) {
file2.Write([]byte(payloadSeparator))
file2.Close()
input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100, 0, false)
input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false)
for i := '1'; i <= '4'; i++ {
msg, _ := input.PluginRead()
@@ -198,7 +198,7 @@ func TestInputFileLoop(t *testing.T) {
file.Write([]byte(payloadSeparator))
file.Close()
input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), true, 100, 0, false)
input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), true)
// Even if we have just 2 requests in file, it should indifinitly loop
for i := 0; i < 1000; i++ {
@@ -226,7 +226,7 @@ func TestInputFileCompressed(t *testing.T) {
name2 := output2.file.Name()
output2.Close()
input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100, 0, false)
input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false)
for i := 0; i < 2000; i++ {
input.PluginRead()
}
@@ -326,7 +326,7 @@ func CreateCaptureFile(requestGenerator *RequestGenerator) *CaptureFile {
func ReadFromCaptureFile(captureFile *os.File, count int, callback writeCallback) (err error) {
wg := new(sync.WaitGroup)
input := NewFileInput(captureFile.Name(), false, 100, 0, false)
input := NewFileInput(captureFile.Name(), false)
output := NewTestOutput(func(msg *Message) {
callback(msg)
wg.Done()
+81 -17
View File
@@ -16,20 +16,53 @@ import (
"github.com/buger/goreplay/tcp"
)
// TCPProtocol is a number to indicate type of protocol
type TCPProtocol uint8
const (
// ProtocolHTTP ...
ProtocolHTTP TCPProtocol = iota
// ProtocolBinary ...
ProtocolBinary
)
// Set is here so that TCPProtocol can implement flag.Var
func (protocol *TCPProtocol) Set(v string) error {
switch v {
case "", "http":
*protocol = ProtocolHTTP
case "binary":
*protocol = ProtocolBinary
default:
return fmt.Errorf("unsupported protocol %s", v)
}
return nil
}
func (protocol *TCPProtocol) String() string {
switch *protocol {
case ProtocolBinary:
return "binary"
case ProtocolHTTP:
return "http"
default:
return ""
}
}
// RAWInputConfig represents configuration that can be applied on raw input
type RAWInputConfig struct {
capture.PcapOptions
Expire time.Duration `json:"input-raw-expire"`
CopyBufferSize size.Size `json:"copy-buffer-size"`
Engine capture.EngineType `json:"input-raw-engine"`
TrackResponse bool `json:"input-raw-track-response"`
Protocol tcp.TCPProtocol `json:"input-raw-protocol"`
RealIPHeader string `json:"input-raw-realip-header"`
Stats bool `json:"input-raw-stats"`
AllowIncomplete bool `json:"input-raw-allow-incomplete"`
quit chan bool // Channel used only to indicate goroutine should shutdown
host string
ports []uint16
Expire time.Duration `json:"input-raw-expire"`
CopyBufferSize size.Size `json:"copy-buffer-size"`
Engine capture.EngineType `json:"input-raw-engine"`
TrackResponse bool `json:"input-raw-track-response"`
Protocol TCPProtocol `json:"input-raw-protocol"`
RealIPHeader string `json:"input-raw-realip-header"`
Stats bool `json:"input-raw-stats"`
quit chan bool // Channel used only to indicate goroutine should shutdown
host string
ports []uint16
}
// RAWInput used for intercepting traffic for given address
@@ -38,7 +71,7 @@ type RAWInput struct {
RAWInputConfig
messageStats []tcp.Stats
listener *capture.Listener
messageParser *tcp.MessageParser
message chan *tcp.Message
cancelListener context.CancelFunc
closed bool
}
@@ -47,6 +80,7 @@ type RAWInput struct {
func NewRAWInput(address string, config RAWInputConfig) (i *RAWInput) {
i = new(RAWInput)
i.RAWInputConfig = config
i.message = make(chan *tcp.Message, 10000)
i.quit = make(chan bool)
host, _ports, err := net.SplitHostPort(address)
@@ -83,12 +117,11 @@ func (i *RAWInput) PluginRead() (*Message, error) {
select {
case <-i.quit:
return nil, ErrorStopped
case msgTCP = <-i.listener.Messages():
case msgTCP = <-i.message:
msg.Data = msgTCP.Data()
}
var msgType byte = ResponsePayload
if msgTCP.Direction == tcp.DirIncoming {
if msgTCP.IsRequest {
msgType = RequestPayload
if i.RealIPHeader != "" {
msg.Data = proto.SetHeader(msg.Data, []byte(i.RealIPHeader), []byte(msgTCP.SrcAddr))
@@ -108,13 +141,14 @@ func (i *RAWInput) PluginRead() (*Message, error) {
stat := msgTCP.Stats
go i.addStats(stat)
}
msgTCP.Finalize()
msgTCP = nil
return &msg, nil
}
func (i *RAWInput) listen(address string) {
var err error
i.listener, err = capture.NewListener(i.host, i.ports, "", i.Engine, i.Protocol, i.TrackResponse, i.Expire, i.AllowIncomplete)
i.listener, err = capture.NewListener(i.host, i.ports, "", i.Engine, i.TrackResponse)
if err != nil {
log.Fatal(err)
}
@@ -123,10 +157,15 @@ func (i *RAWInput) listen(address string) {
if err != nil {
log.Fatal(err)
}
parser := tcp.NewMessageParser(i.CopyBufferSize, i.Expire, Debug, i.messageEmitter)
if i.Protocol == ProtocolHTTP {
parser.Start = http1StartHint
parser.End = http1EndHint
}
var ctx context.Context
ctx, i.cancelListener = context.WithCancel(context.Background())
errCh := i.listener.ListenBackground(ctx)
errCh := i.listener.ListenBackground(ctx, parser.PacketHandler)
<-i.listener.Reading
Debug(1, i)
go func() {
@@ -135,6 +174,10 @@ func (i *RAWInput) listen(address string) {
}()
}
func (i *RAWInput) messageEmitter(m *tcp.Message) {
i.message <- m
}
func (i *RAWInput) String() string {
return fmt.Sprintf("Intercepting traffic from: %s:%s", i.host, strings.Join(strings.Fields(fmt.Sprint(i.ports)), ","))
}
@@ -170,3 +213,24 @@ func (i *RAWInput) addStats(mStats tcp.Stats) {
i.messageStats = append(i.messageStats, mStats)
i.Unlock()
}
func http1StartHint(pckt *tcp.Packet) (isRequest, isResponse bool) {
if proto.HasRequestTitle(pckt.Payload) {
return true, false
}
if proto.HasResponseTitle(pckt.Payload) {
return false, true
}
// No request or response detected
return false, false
}
func http1EndHint(m *tcp.Message) bool {
if m.MissingChunk() {
return false
}
return proto.HasFullPayload(m, m.PacketData()...)
}
+8 -15
View File
@@ -16,7 +16,6 @@ import (
"github.com/buger/goreplay/capture"
"github.com/buger/goreplay/proto"
"github.com/buger/goreplay/tcp"
)
const testRawExpire = time.Millisecond * 200
@@ -44,7 +43,7 @@ func TestRAWInputIPv4(t *testing.T) {
conf := RAWInputConfig{
Engine: capture.EnginePcap,
Expire: 0,
Protocol: tcp.ProtocolHTTP,
Protocol: ProtocolHTTP,
TrackResponse: true,
RealIPHeader: "X-Real-IP",
}
@@ -59,7 +58,6 @@ func TestRAWInputIPv4(t *testing.T) {
} else {
respCounter++
}
wg.Done()
})
@@ -73,18 +71,14 @@ func TestRAWInputIPv4(t *testing.T) {
emitter := NewEmitter()
defer emitter.Close()
go emitter.Start(plugins, Settings.Middleware)
// time.Sleep(time.Second)
for i := 0; i < 1; i++ {
wg.Add(2)
_, err = http.Get(addr)
if err != nil {
t.Error(err)
return
}
}
wg.Wait()
const want = 10
if reqCounter != respCounter && reqCounter != want {
@@ -114,7 +108,7 @@ func TestRAWInputNoKeepAlive(t *testing.T) {
conf := RAWInputConfig{
Engine: capture.EnginePcap,
Expire: testRawExpire,
Protocol: tcp.ProtocolHTTP,
Protocol: ProtocolHTTP,
TrackResponse: true,
}
input := NewRAWInput(":"+port, conf)
@@ -179,7 +173,7 @@ func TestRAWInputIPv6(t *testing.T) {
var respCounter, reqCounter int64
conf := RAWInputConfig{
Engine: capture.EnginePcap,
Protocol: tcp.ProtocolHTTP,
Protocol: ProtocolHTTP,
TrackResponse: true,
}
input := NewRAWInput(originAddr, conf)
@@ -234,11 +228,10 @@ func TestInputRAWChunkedEncoding(t *testing.T) {
originAddr := strings.Replace(origin.Listener.Addr().String(), "[::]", "127.0.0.1", -1)
conf := RAWInputConfig{
Engine: capture.EnginePcap,
Expire: time.Second,
Protocol: tcp.ProtocolHTTP,
TrackResponse: true,
AllowIncomplete: true,
Engine: capture.EnginePcap,
Expire: time.Second,
Protocol: ProtocolHTTP,
TrackResponse: true,
}
input := NewRAWInput(originAddr, conf)
@@ -316,7 +309,7 @@ func BenchmarkRAWInputWithReplay(b *testing.B) {
conf := RAWInputConfig{
Engine: capture.EnginePcap,
Expire: testRawExpire,
Protocol: tcp.ProtocolHTTP,
Protocol: ProtocolHTTP,
TrackResponse: true,
}
input := NewRAWInput(originAddr, conf)
-1
View File
@@ -24,7 +24,6 @@ func (i *DummyOutput) PluginWrite(msg *Message) (int, error) {
n += nn
nn, err = os.Stdout.Write(payloadSeparatorAsBytes)
n += nn
return n, err
}
+13 -33
View File
@@ -7,7 +7,6 @@ import (
"fmt"
"io"
"log"
"math/rand"
"os"
"path/filepath"
"runtime/debug"
@@ -20,21 +19,6 @@ import (
"github.com/buger/goreplay/size"
)
var letters = []rune("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ")
var instanceID string
func init() {
instanceID = randSeq(8)
}
func randSeq(n int) string {
b := make([]rune, n)
for i := range b {
b[i] = letters[rand.Intn(len(letters))]
}
return string(b)
}
var dateFileNameFuncs = map[string]func(*FileOutput) string{
"%Y": func(o *FileOutput) string { return time.Now().Format("2006") },
"%m": func(o *FileOutput) string { return time.Now().Format("01") },
@@ -45,7 +29,6 @@ var dateFileNameFuncs = map[string]func(*FileOutput) string{
"%NS": func(o *FileOutput) string { return fmt.Sprint(time.Now().Nanosecond()) },
"%r": func(o *FileOutput) string { return string(o.currentID) },
"%t": func(o *FileOutput) string { return string(o.payloadType) },
"%i": func(o *FileOutput) string { return instanceID },
}
// FileOutputConfig ...
@@ -62,17 +45,17 @@ type FileOutputConfig struct {
// FileOutput output plugin
type FileOutput struct {
sync.RWMutex
pathTemplate string
currentName string
file *os.File
QueueLength int
writer io.Writer
requestPerFile bool
currentID []byte
payloadType []byte
closed bool
currentFileSize int
totalFileSize size.Size
pathTemplate string
currentName string
file *os.File
QueueLength int
chunkSize int
writer io.Writer
requestPerFile bool
currentID []byte
payloadType []byte
closed bool
totalFileSize size.Size
config *FileOutputConfig
}
@@ -174,7 +157,7 @@ func (o *FileOutput) filename() string {
if o.currentName == "" ||
((o.config.QueueLimit > 0 && o.QueueLength >= o.config.QueueLimit) ||
(o.config.SizeLimit > 0 && o.currentFileSize >= int(o.config.SizeLimit))) {
(o.config.SizeLimit > 0 && o.chunkSize >= int(o.config.SizeLimit))) {
nextChunk = true
}
@@ -253,7 +236,6 @@ func (o *FileOutput) PluginWrite(msg *Message) (n int, err error) {
n += nn
o.totalFileSize += size.Size(n)
o.currentFileSize += n
o.QueueLength++
if Settings.OutputFileConfig.OutputFileMaxSize > 0 && o.totalFileSize >= Settings.OutputFileConfig.OutputFileMaxSize {
@@ -282,7 +264,7 @@ func (o *FileOutput) flush() {
}
if stat, err := o.file.Stat(); err == nil {
o.currentFileSize = int(stat.Size())
o.chunkSize = int(stat.Size())
} else {
Debug(0, "[OUTPUT-HTTP] error accessing file size", err)
}
@@ -308,8 +290,6 @@ func (o *FileOutput) closeLocked() error {
}
o.closed = true
o.currentFileSize = 0
return nil
}
+1 -1
View File
@@ -39,7 +39,7 @@ func TestFileOutput(t *testing.T) {
emitter.Close()
var counter int64
input2 := NewFileInput("/tmp/test_requests.gor", false, 100, 0, false)
input2 := NewFileInput("/tmp/test_requests.gor", false)
output2 := NewTestOutput(func(*Message) {
atomic.AddInt64(&counter, 1)
wg.Done()
-1
View File
@@ -215,7 +215,6 @@ func (o *HTTPOutput) sendRequest(client *HTTPClient, msg *Message) {
if !isRequestPayload(msg.Meta) {
return
}
uuid := payloadID(msg.Meta)
start := time.Now()
resp, err := client.Send(msg.Data)
+4 -3
View File
@@ -95,15 +95,16 @@ func (o *TCPOutput) worker(bufferIndex int) {
}
}
func (o *TCPOutput) getBufferIndex(msg *Message) int {
func (o *TCPOutput) getBufferIndex(data []byte) int {
if !o.config.Sticky {
o.workerIndex++
return int(o.workerIndex) % o.config.Workers
}
hasher := fnv.New32a()
hasher.Write(payloadID(msg.Meta))
hasher.Write(payloadMeta(data)[1])
return int(hasher.Sum32()) % o.config.Workers
}
// PluginWrite writes message to this plugin
@@ -112,7 +113,7 @@ func (o *TCPOutput) PluginWrite(msg *Message) (n int, err error) {
return len(msg.Data), nil
}
bufferIndex := o.getBufferIndex(msg)
bufferIndex := o.getBufferIndex(msg.Data)
o.buf[bufferIndex] <- msg
if Settings.OutputTCPStats {
+4 -5
View File
@@ -125,9 +125,8 @@ func TestBufferDistribution(t *testing.T) {
}
}
func getTestBytes() *Message {
return &Message{
Meta: payloadHeader(RequestPayload, uuid(), time.Now().UnixNano(), -1),
Data: []byte("GET / HTTP/1.1\r\nHost: www.w3.org\r\nUser-Agent: Go 1.1 package http\r\nAccept-Encoding: gzip\r\n\r\n"),
}
func getTestBytes() []byte {
reqh := payloadHeader(RequestPayload, uuid(), time.Now().UnixNano(), -1)
reqb := append(reqh, []byte("GET / HTTP/1.1\r\nHost: www.w3.org\r\nUser-Agent: Go 1.1 package http\r\nAccept-Encoding: gzip\r\n\r\n")...)
return reqb
}
+2 -1
View File
@@ -83,6 +83,7 @@ func (plugins *InOutPlugins) registerPlugin(constructor interface{}, options ...
plugins.Outputs = append(plugins.Outputs, w)
}
plugins.All = append(plugins.All, plugin)
}
// NewPlugins specify and initialize all available plugins
@@ -118,7 +119,7 @@ func NewPlugins() *InOutPlugins {
}
for _, options := range Settings.InputFile {
plugins.registerPlugin(NewFileInput, options, Settings.InputFileLoop, Settings.InputFileReadDepth, Settings.InputFileMaxWait, Settings.InputFileDryRun)
plugins.registerPlugin(NewFileInput, options, Settings.InputFileLoop)
}
for _, path := range Settings.OutputFile {
+28 -43
View File
@@ -77,12 +77,9 @@ func header(payload []byte, name []byte) (value []byte, headerStart, headerEnd,
headerEnd += headerStart
colonIndex = bytes.IndexByte(payload[headerStart:headerEnd], ':')
if colonIndex == -1 {
// Malformed header, skip, most likely packet with partial headers
headerStart = headerEnd + 1
continue
break
}
colonIndex += headerStart
if bytes.EqualFold(payload[headerStart:colonIndex], name) {
valueStart = colonIndex + 1
valueEnd = headerEnd - 2
@@ -451,36 +448,34 @@ type ProtocolStateSetter interface {
ProtocolState() interface{}
}
type HTTPState struct {
Body int // body index
HeaderStart int
HeaderEnd int
HeaderParsed bool // we checked necessary headers
HasFullPayload bool // all chunks has been parsed
IsChunked bool // Transfer-Encoding: chunked
BodyLen int // Content-Length's value
HasTrailer bool // Trailer header?
Continue100 bool
type httpProto struct {
body int // body index
headerStart int
headerParsed bool // we checked necessary headers
hasFullBody bool // all chunks has been parsed
isChunked bool // Transfer-Encoding: chunked
bodyLen int // Content-Length's value
hasTrailer bool // Trailer header?
}
// HasFullPayload checks if this message has full or valid payloads and returns true.
// Message param is optional but recommended on cases where 'data' is storing
// partial-to-full stream of bytes(packets).
func HasFullPayload(m ProtocolStateSetter, payloads ...[]byte) bool {
var state *HTTPState
var state *httpProto
if m != nil {
state, _ = m.ProtocolState().(*HTTPState)
state, _ = m.ProtocolState().(*httpProto)
}
if state == nil {
state = new(HTTPState)
state = new(httpProto)
if m != nil {
m.SetProtocolState(state)
}
}
if state.HeaderStart < 1 {
if state.headerStart < 1 {
for _, data := range payloads {
state.HeaderStart = MIMEHeadersStartPos(data)
if state.HeaderStart < 0 {
state.headerStart = MIMEHeadersStartPos(data)
if state.headerStart < 0 {
return false
} else {
break
@@ -488,7 +483,7 @@ func HasFullPayload(m ProtocolStateSetter, payloads ...[]byte) bool {
}
}
if state.Body < 1 || state.HeaderEnd < 1 {
if state.body < 1 {
var pos int
for _, data := range payloads {
endPos := MIMEHeadersEndPos(data)
@@ -496,42 +491,32 @@ func HasFullPayload(m ProtocolStateSetter, payloads ...[]byte) bool {
pos += len(data)
} else {
pos += endPos
state.HeaderEnd = pos
}
if endPos > 0 {
state.Body = pos
state.body = pos
break
}
}
}
if state.HeaderEnd < 1 {
return false
}
if !state.HeaderParsed {
if !state.headerParsed {
var pos int
for _, data := range payloads {
chunked := Header(data, []byte("Transfer-Encoding"))
if len(chunked) > 0 && bytes.Index(data, []byte("chunked")) > 0 {
state.IsChunked = true
state.isChunked = true
// trailers are generally not allowed in non-chunks body
state.HasTrailer = len(Header(data, []byte("Trailer"))) > 0
state.hasTrailer = len(Header(data, []byte("Trailer"))) > 0
} else {
contentLen := Header(data, []byte("Content-Length"))
state.BodyLen, _ = atoI(contentLen, 10)
state.bodyLen, _ = atoI(contentLen, 10)
}
pos += len(data)
if string(Header(data, []byte("Expect"))) == "100-continue" {
state.Continue100 = true
}
if state.BodyLen > 0 || pos >= state.Body {
state.HeaderParsed = true
if state.bodyLen > 0 || pos >= state.body {
state.headerParsed = true
break
}
}
@@ -541,22 +526,22 @@ func HasFullPayload(m ProtocolStateSetter, payloads ...[]byte) bool {
for _, data := range payloads {
bodyLen += len(data)
}
bodyLen -= state.Body
bodyLen -= state.body
if state.IsChunked {
if state.isChunked {
// check chunks
if bodyLen < 1 {
return false
}
// check trailer headers
if state.HasTrailer {
if state.hasTrailer {
if bytes.HasSuffix(payloads[len(payloads)-1], []byte("\r\n\r\n")) {
return true
}
} else {
if bytes.HasSuffix(payloads[len(payloads)-1], []byte("0\r\n\r\n")) {
state.HasFullPayload = true
state.hasFullBody = true
return true
}
}
@@ -565,7 +550,7 @@ func HasFullPayload(m ProtocolStateSetter, payloads ...[]byte) bool {
}
// check for content-length header
return state.BodyLen == bodyLen
return state.bodyLen == bodyLen
}
// this works with positive integers
-6
View File
@@ -56,12 +56,6 @@ func TestHeader(t *testing.T) {
if val = Header(payload, []byte("host")); !bytes.Equal(val, []byte("www.w3.org")) {
t.Error("Should find lower case 1 word header")
}
payload = []byte("GT\r\nContent-Length: 10\r\n\r\n")
if val = Header(payload, []byte("Content-Length")); !bytes.Equal(val, []byte("10")) {
t.Error("Should find in partial payload")
}
}
func TestMIMEHeadersEndPos(t *testing.T) {
-205
View File
@@ -1,205 +0,0 @@
package ring
import (
"errors"
"runtime"
"sync/atomic"
"time"
)
var (
// ErrDisposed is returned when an operation is performed on a disposed
// queue.
ErrDisposed = errors.New(`queue: disposed`)
// ErrTimeout is returned when an applicable queue operation times out.
ErrTimeout = errors.New(`queue: poll timed out`)
// ErrEmptyQueue is returned when an non-applicable queue operation was called
// due to the queue's empty item state
ErrEmptyQueue = errors.New(`queue: empty queue`)
)
// roundUp takes a uint64 greater than 0 and rounds it up to the next
// power of 2.
func roundUp(v uint64) uint64 {
v--
v |= v >> 1
v |= v >> 2
v |= v >> 4
v |= v >> 8
v |= v >> 16
v |= v >> 32
v++
return v
}
type node struct {
position uint64
data interface{}
}
type nodes []node
// RingBuffer is a MPMC buffer that achieves threadsafety with CAS operations
// only. A put on full or get on empty call will block until an item
// is put or retrieved. Calling Dispose on the RingBuffer will unblock
// any blocked threads with an error. This buffer is similar to the buffer
// described here: http://www.1024cores.net/home/lock-free-algorithms/queues/bounded-mpmc-queue
// with some minor additions.
type RingBuffer struct {
_padding0 [8]uint64
queue uint64
_padding1 [8]uint64
dequeue uint64
_padding2 [8]uint64
mask, disposed uint64
_padding3 [8]uint64
nodes nodes
}
func (rb *RingBuffer) init(size uint64) {
size = roundUp(size)
rb.nodes = make(nodes, size)
for i := uint64(0); i < size; i++ {
rb.nodes[i] = node{position: i}
}
rb.mask = size - 1 // so we don't have to do this with every put/get operation
}
// Put adds the provided item to the queue. If the queue is full, this
// call will block until an item is added to the queue or Dispose is called
// on the queue. An error will be returned if the queue is disposed.
func (rb *RingBuffer) Put(item interface{}) error {
_, err := rb.put(item, false)
return err
}
// Offer adds the provided item to the queue if there is space. If the queue
// is full, this call will return false. An error will be returned if the
// queue is disposed.
func (rb *RingBuffer) Offer(item interface{}) (bool, error) {
return rb.put(item, true)
}
func (rb *RingBuffer) put(item interface{}, offer bool) (bool, error) {
var n *node
pos := atomic.LoadUint64(&rb.queue)
L:
for {
if atomic.LoadUint64(&rb.disposed) == 1 {
return false, ErrDisposed
}
n = &rb.nodes[pos&rb.mask]
seq := atomic.LoadUint64(&n.position)
switch dif := seq - pos; {
case dif == 0:
if atomic.CompareAndSwapUint64(&rb.queue, pos, pos+1) {
break L
}
case dif < 0:
panic(`Ring buffer in a compromised state during a put operation.`)
default:
pos = atomic.LoadUint64(&rb.queue)
}
if offer {
return false, nil
}
runtime.Gosched() // free up the cpu before the next iteration
}
n.data = item
atomic.StoreUint64(&n.position, pos+1)
return true, nil
}
// Get will return the next item in the queue. This call will block
// if the queue is empty. This call will unblock when an item is added
// to the queue or Dispose is called on the queue. An error will be returned
// if the queue is disposed.
func (rb *RingBuffer) Get() (interface{}, error) {
return rb.Poll(0)
}
// Poll will return the next item in the queue. This call will block
// if the queue is empty. This call will unblock when an item is added
// to the queue, Dispose is called on the queue, or the timeout is reached. An
// error will be returned if the queue is disposed or a timeout occurs. A
// non-positive timeout will block indefinitely.
func (rb *RingBuffer) Poll(timeout time.Duration) (interface{}, error) {
var (
n *node
pos = atomic.LoadUint64(&rb.dequeue)
start time.Time
)
if timeout > 0 {
start = time.Now()
}
L:
for {
if atomic.LoadUint64(&rb.disposed) == 1 {
return nil, ErrDisposed
}
n = &rb.nodes[pos&rb.mask]
seq := atomic.LoadUint64(&n.position)
switch dif := seq - (pos + 1); {
case dif == 0:
if atomic.CompareAndSwapUint64(&rb.dequeue, pos, pos+1) {
break L
}
case dif < 0:
panic(`Ring buffer in compromised state during a get operation.`)
default:
pos = atomic.LoadUint64(&rb.dequeue)
}
if timeout > 0 && time.Since(start) >= timeout {
return nil, ErrTimeout
}
if timeout < 0 {
return nil, ErrTimeout
}
runtime.Gosched() // free up the cpu before the next iteration
}
data := n.data
n.data = nil
atomic.StoreUint64(&n.position, pos+rb.mask+1)
return data, nil
}
// Len returns the number of items in the queue.
func (rb *RingBuffer) Len() uint64 {
return atomic.LoadUint64(&rb.queue) - atomic.LoadUint64(&rb.dequeue)
}
// Cap returns the capacity of this ring buffer.
func (rb *RingBuffer) Cap() uint64 {
return uint64(len(rb.nodes))
}
// Dispose will dispose of this queue and free any blocked threads
// in the Put and/or Get methods. Calling those methods on a disposed
// queue will return an error.
func (rb *RingBuffer) Dispose() {
atomic.CompareAndSwapUint64(&rb.disposed, 0, 1)
}
// IsDisposed will return a bool indicating if this queue has been
// disposed.
func (rb *RingBuffer) IsDisposed() bool {
return atomic.LoadUint64(&rb.disposed) == 1
}
// NewRingBuffer will allocate, initialize, and return a ring buffer
// with the specified size.
func NewRingBuffer(size uint64) *RingBuffer {
rb := &RingBuffer{}
rb.init(size)
return rb
}
+1 -1
View File
@@ -127,7 +127,7 @@ func TestInputFileFromS3(t *testing.T) {
<-output.closeCh
}
input := NewFileInput(fmt.Sprintf("s3://test-gor-eu/%d", rnd), false, 100, 0, false)
input := NewFileInput(fmt.Sprintf("s3://test-gor-eu/%d", rnd), false)
buf := make([]byte, 1000)
for i := 0; i <= 19999; i++ {
+5 -12
View File
@@ -45,13 +45,10 @@ type AppSettings struct {
OutputTCPConfig TCPOutputConfig
OutputTCPStats bool `json:"output-tcp-stats"`
InputFile MultiOption `json:"input-file"`
InputFileLoop bool `json:"input-file-loop"`
InputFileReadDepth int `json:"input-file-read-depth"`
InputFileDryRun bool `json:"input-file-dry-run"`
InputFileMaxWait time.Duration `json:"input-file-max-wait"`
OutputFile MultiOption `json:"output-file"`
OutputFileConfig FileOutputConfig
InputFile MultiOption `json:"input-file"`
InputFileLoop bool `json:"input-file-loop"`
OutputFile MultiOption `json:"output-file"`
OutputFileConfig FileOutputConfig
InputRAW MultiOption `json:"input_raw"`
RAWInputConfig
@@ -116,9 +113,6 @@ func init() {
flag.Var(&Settings.InputFile, "input-file", "Read requests from file: \n\tgor --input-file ./requests.gor --output-http staging.com")
flag.BoolVar(&Settings.InputFileLoop, "input-file-loop", false, "Loop input files, useful for performance testing.")
flag.IntVar(&Settings.InputFileReadDepth, "input-file-read-depth", 100, "GoReplay tries to read and cache multiple records, in advance. In parallel it also perform sorting of requests, if they came out of order. Since it needs hold this buffer in memory, bigger values can cause worse performance")
flag.BoolVar(&Settings.InputFileDryRun, "input-file-dry-run", false, "Simulate reading from the data source without replaying it. You will get information about expected replay time, number of found records etc.")
flag.DurationVar(&Settings.InputFileMaxWait, "input-file-max-wait", 0, "Set the maximum time between requests. Can help in situations when you have too long periods between request, and you want to skip them. Example: --input-raw-max-wait 1s")
flag.Var(&Settings.OutputFile, "output-file", "Write incoming requests to file: \n\tgor --input-raw :80 --output-file ./requests.gor")
flag.DurationVar(&Settings.OutputFileConfig.FlushInterval, "output-file-flush-interval", time.Second, "Interval for forcing buffer flush to the file, default: 1s.")
@@ -147,7 +141,6 @@ func init() {
flag.BoolVar(&Settings.Promiscuous, "input-raw-promisc", false, "enable promiscuous mode")
flag.BoolVar(&Settings.Monitor, "input-raw-monitor", false, "enable RF monitor mode")
flag.BoolVar(&Settings.Stats, "input-raw-stats", false, "enable stats generator on raw TCP messages")
flag.BoolVar(&Settings.AllowIncomplete, "input-raw-allow-incomplete", false, "If turned on Gor will record HTTP messages with missing packets")
flag.StringVar(&Settings.Middleware, "middleware", "", "Used for modifying traffic using external command")
@@ -194,7 +187,7 @@ func init() {
flag.StringVar(&Settings.KafkaTLSConfig.ClientCert, "kafka-tls-client-cert", "", "Client certificate for Kafka TLS Config (mandatory with to kafka-tls-ca-cert and kafka-tls-client-key)")
flag.StringVar(&Settings.KafkaTLSConfig.ClientKey, "kafka-tls-client-key", "", "Client Key for Kafka TLS Config (mandatory with to kafka-tls-client-cert and kafka-tls-client-key)")
flag.Var(&Settings.ModifierConfig.Headers, "http-set-header", "Inject additional headers to http request:\n\tgor --input-raw :8080 --output-http staging.com --http-set-header 'User-Agent: Gor'")
flag.Var(&Settings.ModifierConfig.Headers, "http-set-header", "Inject additional headers to http reqest:\n\tgor --input-raw :8080 --output-http staging.com --http-set-header 'User-Agent: Gor'")
flag.Var(&Settings.ModifierConfig.HeaderRewrite, "http-rewrite-header", "Rewrite the request header based on a mapping:\n\tgor --input-raw :8080 --output-http staging.com --http-rewrite-header Host: (.*).example.com,$1.beta.example.com")
flag.Var(&Settings.ModifierConfig.Params, "http-set-param", "Set request url param, if param already exists it will be overwritten:\n\tgor --input-raw :8080 --output-http staging.com --http-set-param api_key=1")
flag.Var(&Settings.ModifierConfig.Methods, "http-allow-method", "Whitelist of HTTP methods to replay. Anything else will be dropped:\n\tgor --input-raw :8080 --output-http staging.com --http-allow-method GET --http-allow-method OPTIONS")
-17
View File
@@ -1,17 +0,0 @@
package simpletime
import (
"time"
)
var Now time.Time
func init() {
go func() {
for {
// Accurate enough
Now = time.Now()
time.Sleep(100 * time.Millisecond)
}
}()
}
+94 -205
View File
@@ -3,50 +3,13 @@ package tcp
import (
"encoding/binary"
"encoding/hex"
"fmt"
"net"
"reflect"
_ "fmt"
"sort"
"time"
"unsafe"
"github.com/buger/goreplay/proto"
"github.com/buger/goreplay/size"
)
// TCPProtocol is a number to indicate type of protocol
type TCPProtocol uint8
const (
// ProtocolHTTP ...
ProtocolHTTP TCPProtocol = iota
// ProtocolBinary ...
ProtocolBinary
)
// Set is here so that TCPProtocol can implement flag.Var
func (protocol *TCPProtocol) Set(v string) error {
switch v {
case "", "http":
*protocol = ProtocolHTTP
case "binary":
*protocol = ProtocolBinary
default:
return fmt.Errorf("unsupported protocol %s", v)
}
return nil
}
func (protocol *TCPProtocol) String() string {
switch *protocol {
case ProtocolBinary:
return "binary"
case ProtocolHTTP:
return "http"
default:
return ""
}
}
// Stats every message carry its own stats object
type Stats struct {
LostData int
@@ -55,7 +18,7 @@ type Stats struct {
End time.Time // last packet's timestamp
SrcAddr string
DstAddr string
Direction Dir
IsRequest bool
TimedOut bool // timeout before getting the whole message
Truncated bool // last packet truncated due to max message size
IPversion byte
@@ -75,7 +38,7 @@ func (m *Message) UUID() []byte {
pckt := m.packets[0]
// check if response or request have generated the ID before.
if m.Direction == DirIncoming {
if m.IsRequest {
streamID = uint64(pckt.SrcPort)<<48 | uint64(pckt.DstPort)<<32 |
uint64(ip2int(pckt.SrcIP))
} else {
@@ -86,7 +49,7 @@ func (m *Message) UUID() []byte {
id := make([]byte, 12)
binary.BigEndian.PutUint64(id, streamID)
if m.Direction == DirIncoming {
if m.IsRequest {
binary.BigEndian.PutUint32(id[8:], pckt.Ack)
} else {
binary.BigEndian.PutUint32(id[8:], pckt.Seq)
@@ -98,11 +61,13 @@ func (m *Message) UUID() []byte {
return uuidHex
}
func (m *Message) add(packet *Packet) bool {
func (m *Message) add(packet *Packet) {
// fmt.Println("SEQ:", packet.Seq, " - ", len(packet.Payload))
// Skip duplicates
for _, p := range m.packets {
if p.Seq == packet.Seq {
return false
return
}
}
@@ -126,8 +91,6 @@ func (m *Message) add(packet *Packet) bool {
if packet.Timestamp.After(m.End) || m.End.IsZero() {
m.End = packet.Timestamp
}
return true
}
// Packets returns packets of the message
@@ -161,18 +124,15 @@ func (m *Message) PacketData() [][]byte {
// Data returns data in this message
func (m *Message) Data() []byte {
packetData := m.PacketData()
tmp := packetData[0]
if len(packetData) > 0 {
tmp, _ = copySlice(tmp, len(packetData[0]), packetData[1:]...)
var totalLen int
for _, p := range m.packets {
totalLen += len(p.Payload)
}
tmp := make([]byte, totalLen)
// Remove Expect header, since its replay not fully supported
if state, ok := m.feedback.(*proto.HTTPState); ok {
if state.Continue100 {
tmp = proto.DeleteHeader(tmp, []byte("Expect"))
}
var i int
for _, p := range m.packets {
i += copy(tmp[i:], p.Payload)
}
return tmp
@@ -193,9 +153,20 @@ func (m *Message) Sort() {
sort.SliceStable(m.packets, func(i, j int) bool { return m.packets[i].Seq < m.packets[j].Seq })
}
func (m *Message) Finalize() {
// Allow re-use memory
for _, p := range m.packets {
packetPool.Put(p)
}
}
// Emitter message handler
type Emitter func(*Message)
// Debugger is the debugger function. first params is the indicator of the issue's priority
// the higher the number, the lower the priority. it can be 4 <= level <= 6.
type Debugger func(int, ...interface{})
// HintEnd hints the parser to stop the session, see MessageParser.End
// when set, it will be executed before checking FIN or RST flag
type HintEnd func(*Message) bool
@@ -207,65 +178,54 @@ type HintStart func(*Packet) (IsRequest, IsOutgoing bool)
// MessageParser holds data of all tcp messages in progress(still receiving/sending packets).
// message is identified by its source port and dst port, and last 4bytes of src IP.
type MessageParser struct {
m map[uint64]*Message
messageExpire time.Duration // the maximum time to wait for the final packet, minimum is 100ms
allowIncompete bool
End HintEnd
Start HintStart
ticker *time.Ticker
messages chan *Message
packets chan *PcapPacket
close chan struct{} // to signal that we are able to close
ports []uint16
ips []net.IP
debug Debugger
maxSize size.Size // maximum message size, default 5mb
m map[uint64]*Message
emit Emitter
messageExpire time.Duration // the maximum time to wait for the final packet, minimum is 100ms
End HintEnd
Start HintStart
ticker *time.Ticker
packets chan *Packet
msgs int32 // messages in the parser
close chan struct{} // to signal that we are able to close
}
// NewMessageParser returns a new instance of message parser
func NewMessageParser(messages chan *Message, ports []uint16, ips []net.IP, messageExpire time.Duration, allowIncompete bool) (parser *MessageParser) {
func NewMessageParser(maxSize size.Size, messageExpire time.Duration, debugger Debugger, emitHandler Emitter) (parser *MessageParser) {
parser = new(MessageParser)
parser.messageExpire = messageExpire
if parser.messageExpire == 0 {
parser.messageExpire = time.Millisecond * 1000
parser.debug = debugger
parser.emit = emitHandler
parser.messageExpire = time.Millisecond * 100
if parser.messageExpire < messageExpire {
parser.messageExpire = messageExpire
}
parser.allowIncompete = allowIncompete
parser.packets = make(chan *PcapPacket, 10000)
if messages == nil {
messages = make(chan *Message, 1000)
parser.maxSize = maxSize
if parser.maxSize < 1 {
parser.maxSize = 5 << 20
}
parser.messages = messages
parser.packets = make(chan *Packet, 1000)
parser.m = make(map[uint64]*Message)
parser.ticker = time.NewTicker(time.Millisecond * 100)
parser.ticker = time.NewTicker(time.Millisecond * 50)
parser.close = make(chan struct{}, 1)
parser.ports = ports
parser.ips = ips
go parser.wait()
return parser
}
var packetLen int
// Packet returns packet handler
func (parser *MessageParser) PacketHandler(packet *PcapPacket) {
packetLen++
func (parser *MessageParser) PacketHandler(packet *Packet) {
parser.packets <- packet
}
func (parser *MessageParser) wait() {
var (
now time.Time
pckt *Packet
now time.Time
)
for {
select {
case pckt := <-parser.packets:
parser.processPacket(parser.parsePacket(pckt))
case pckt = <-parser.packets:
parser.processPacket(pckt)
case now = <-parser.ticker.C:
parser.timer(now)
case <-parser.close:
@@ -273,157 +233,86 @@ func (parser *MessageParser) wait() {
// parser.Close should wait for this function to return
parser.close <- struct{}{}
return
// default:
}
}
}
func (parser *MessageParser) parsePacket(pcapPkt *PcapPacket) *Packet {
pckt, err := ParsePacket(pcapPkt.Data, pcapPkt.LType, pcapPkt.LTypeLen, pcapPkt.Ci, false)
if err != nil {
if _, empty := err.(EmptyPacket); !empty {
stats.Add("packet_error", 1)
}
return nil
}
for _, p := range parser.ports {
if pckt.DstPort == p {
for _, ip := range parser.ips {
if pckt.DstIP.Equal(ip) {
pckt.Direction = DirIncoming
break
}
}
break
}
}
return pckt
}
func (parser *MessageParser) processPacket(pckt *Packet) {
if pckt == nil {
return
}
var in, out bool
// Trying to build unique hash, but there is small chance of collision
// No matter if it is request or response, all packets in the same message have same
m, ok := parser.m[pckt.MessageID()]
switch {
case ok:
if m.Direction == DirUnknown {
if in, out := parser.Start(pckt); in || out {
if in {
m.Direction = DirIncoming
} else {
m.Direction = DirOutcoming
}
}
}
parser.addPacket(m, pckt)
return
case pckt.Direction == DirUnknown && parser.Start != nil:
if in, out := parser.Start(pckt); in || out {
if in {
pckt.Direction = DirIncoming
} else {
pckt.Direction = DirOutcoming
case parser.Start != nil:
if in, out = parser.Start(pckt); !(in || out) {
// Packet can be received out of order, so give it another chance
if pckt.Retry < 1 && len(pckt.Payload) > 0 {
// Requeue not known packets
pckt.Retry++
select {
case parser.packets <- pckt:
default:
packetPool.Put(pckt)
// fmt.Println("Skipping packet")
}
}
return
}
}
m = new(Message)
m.Direction = pckt.Direction
m.SrcAddr = pckt.SrcIP.String()
m.DstAddr = pckt.DstIP.String()
m.IsRequest = in
parser.m[pckt.MessageID()] = m
m.Start = pckt.Timestamp
m.parser = parser
parser.addPacket(m, pckt)
}
func (parser *MessageParser) addPacket(m *Message, pckt *Packet) bool {
if !m.add(pckt) {
return false
func (parser *MessageParser) addPacket(m *Message, pckt *Packet) {
trunc := m.Length + len(pckt.Payload) - int(parser.maxSize)
if trunc > 0 {
m.Truncated = true
pckt.Payload = pckt.Payload[:int(parser.maxSize)-m.Length]
}
m.add(pckt)
switch {
// if one of this cases matches, we dispatch the message
case trunc >= 0:
case parser.End != nil && parser.End(m):
default:
// continue to receive packets
return
}
// If we are using protocol parsing, like HTTP, depend on its parsing func.
// For the binary procols wait for message to expire
if parser.End != nil {
if parser.End(m) {
parser.Emit(m)
return true
}
parser.Fix100Continue(m)
}
return true
}
func (parser *MessageParser) Fix100Continue(m *Message) {
if state, ok := m.feedback.(*proto.HTTPState); ok && state.Continue100 {
delete(parser.m, m.packets[0].MessageID())
// Shift Ack by given offset
// Size of "HTTP/1.1 100 Continue\r\n\r\n" message
for _, p := range m.packets {
p.messageID = 0
p.Ack += 25
}
// If next section was aready approved and received, merge messages
if next, found := parser.m[m.packets[0].MessageID()]; found {
for _, p := range next.packets {
parser.addPacket(m, p)
}
}
// Re-add (or override) again with new message and ID
parser.m[m.packets[0].MessageID()] = m
}
}
func (parser *MessageParser) Read() *Message {
m := <-parser.messages
return m
parser.Emit(m)
}
func (parser *MessageParser) Emit(m *Message) {
stats.Add("message_count", 1)
delete(parser.m, m.packets[0].MessageID())
parser.messages <- m
parser.emit(m)
}
func GetUnexportedField(field reflect.Value) interface{} {
return reflect.NewAt(field.Type(), unsafe.Pointer(field.UnsafeAddr())).Elem().Interface()
}
var failMsg int
func (parser *MessageParser) timer(now time.Time) {
packetLen = 0
packetQueueLen.Set(int64(len(parser.packets)))
messageQueueLen.Set(int64(len(parser.m)))
for _, m := range parser.m {
if now.Sub(m.End) > parser.messageExpire {
m.TimedOut = true
stats.Add("message_timeout_count", 1)
failMsg++
if parser.End == nil || parser.allowIncompete {
parser.Emit(m)
}
delete(parser.m, m.packets[0].MessageID())
parser.Emit(m)
}
}
}
// this function should not block other parser operations
func (parser *MessageParser) Debug(level int, args ...interface{}) {
if parser.debug != nil {
parser.debug(level, args...)
}
}
func (parser *MessageParser) Close() error {
parser.close <- struct{}{}
<-parser.close // wait for timer to be closed!
+30 -96
View File
@@ -2,54 +2,28 @@ package tcp
import (
"encoding/binary"
"expvar"
"fmt"
"net"
"sync"
"time"
"github.com/google/gopacket"
)
func copySlice(to []byte, skip int, from ...[]byte) ([]byte, int) {
var totalLen int
for _, s := range from {
totalLen += len(s)
func copySlice(b, a []byte) []byte {
if cap(b) < len(a) {
b = make([]byte, len(a))
}
totalLen += skip
if cap(to) < totalLen {
diff := totalLen - cap(to)
to = append(to, make([]byte, diff)...)
}
for _, s := range from {
skip += copy(to[skip:], s)
}
return to, skip
copy(b, a)
return b[:len(a)]
}
var stats *expvar.Map
var packetQueueLen, messageQueueLen *expvar.Int
func init() {
packetQueueLen = new(expvar.Int)
messageQueueLen = new(expvar.Int)
stats = expvar.NewMap("tcp")
stats.Init()
stats.Set("packet_queue", packetQueueLen)
stats.Set("message_queue", messageQueueLen)
var packetPool = sync.Pool{
New: func() interface{} {
return new(Packet)
},
}
type Dir int
const (
DirUnknown = iota
DirIncoming
DirOutcoming
)
/*
Packet represent data and layers of packet.
parser extracts information from pcap Packet. functions of *Packet doesn't validate if packet is nil,
@@ -57,7 +31,6 @@ calllers must make sure that ParsePacket has'nt returned any error before callin
function.
*/
type Packet struct {
Direction Dir
messageID uint64
SrcIP, DstIP net.IP
Version uint8
@@ -66,45 +39,24 @@ type Packet struct {
ACK, SYN, FIN, RST bool
Lost uint32
Retry int
CaptureLength int
Timestamp time.Time
Payload []byte
buf []byte
created time.Time
gc bool
}
type PcapPacket struct {
Data []byte
LType int
LTypeLen int
Ci *gopacket.CaptureInfo
}
// ParsePacket parse raw packets
func ParsePacket(data []byte, lType, lTypeLen int, ci *gopacket.CaptureInfo, allowEmpty bool) (pckt *Packet, err error) {
pckt = new(Packet)
if err := pckt.parse(data, lType, lTypeLen, ci, allowEmpty); err != nil {
return nil, err
}
return pckt, nil
}
func (pckt *Packet) parse(data []byte, lType, lTypeLen int, cp *gopacket.CaptureInfo, allowEmpty bool) error {
func ParsePacket(data []byte, lType, lTypeLen int, cp *gopacket.CaptureInfo) (pckt *Packet, err error) {
pckt = packetPool.Get().(*Packet)
pckt.Retry = 0
pckt.messageID = 0
pckt.buf = pckt.buf[:]
// TODO: check resolution
pckt.Timestamp = cp.Timestamp
if len(data) < lTypeLen {
return ErrHdrLength("Link")
return nil, ErrHdrLength("Link")
}
if len(data) <= lTypeLen {
return ErrHdrMissing("IPv4 or IPv6")
return nil, ErrHdrMissing("IPv4 or IPv6")
}
ldata := data[lTypeLen:]
@@ -114,72 +66,63 @@ func (pckt *Packet) parse(data []byte, lType, lTypeLen int, cp *gopacket.Capture
if ldata[0]>>4 == 4 {
// IPv4 header
if len(ldata) < 20 {
return ErrHdrLength("IPv4")
return nil, ErrHdrLength("IPv4")
}
proto = ldata[9]
ihl := int(ldata[0]&0x0F) * 4
if ihl < 20 {
return ErrHdrInvalid("IPv4's IHL")
return nil, ErrHdrInvalid("IPv4's IHL")
}
if len(ldata) < ihl {
return ErrHdrLength("IPv4 opts")
return nil, ErrHdrLength("IPv4 opts")
}
netLayer = ldata[:ihl]
} else if ldata[0]>>4 == 6 {
if len(ldata) < 40 {
return ErrHdrLength("IPv6")
return nil, ErrHdrLength("IPv6")
}
proto = ldata[6]
totalLen := 40
for ipv6ExtensionHdr(proto) {
hdr := len(ldata) - totalLen
if hdr < 8 {
return ErrHdrExpected("IPv6 opts")
return nil, ErrHdrExpected("IPv6 opts")
}
extLen := 8
if proto != 44 {
extLen = int(ldata[totalLen+1]+1) * 8
}
if hdr < extLen {
return ErrHdrLength("IPv6 opts")
return nil, ErrHdrLength("IPv6 opts")
}
proto = ldata[totalLen]
totalLen += extLen
}
netLayer = ldata[:totalLen]
} else {
return ErrHdrExpected("IPv4 or IPv6")
return nil, ErrHdrExpected("IPv4 or IPv6")
}
if proto != 6 {
return ErrHdrExpected("TCP")
return nil, ErrHdrExpected("TCP")
}
if len(data) <= len(netLayer) {
return ErrHdrMissing("TCP")
return nil, ErrHdrMissing("TCP")
}
ndata := ldata[len(netLayer):]
// TCP header
if len(ndata) < 20 {
return ErrHdrLength("TCP")
return nil, ErrHdrLength("TCP")
}
dOf := int(ndata[12]>>4) * 4
if dOf < 20 {
return ErrHdrInvalid("TCP's ndata offset")
return nil, ErrHdrInvalid("TCP's ndata offset")
}
if len(ndata) < dOf {
return ErrHdrLength("TCP opts")
return nil, ErrHdrLength("TCP opts")
}
// There are case when packet have padding but dOf shows its not
empty := true
for i := 0; i < len(ndata[dOf:]); i++ {
if ndata[dOf:][i] != 0 {
empty = false
break
}
}
if !allowEmpty && empty {
return EmptyPacket("")
if len(ndata[dOf:]) == 0 {
return nil, fmt.Errorf("Packet without Data")
}
if (netLayer[0] >> 4) == 4 {
@@ -196,7 +139,6 @@ func (pckt *Packet) parse(data []byte, lType, lTypeLen int, cp *gopacket.Capture
transLayer = ndata[:dOf]
pckt.CaptureLength = cp.CaptureLength
pckt.SrcPort = binary.BigEndian.Uint16(transLayer[0:2])
pckt.DstPort = binary.BigEndian.Uint16(transLayer[2:4])
pckt.Seq = binary.BigEndian.Uint32(transLayer[4:8])
@@ -206,10 +148,8 @@ func (pckt *Packet) parse(data []byte, lType, lTypeLen int, cp *gopacket.Capture
pckt.RST = transLayer[13]&0x04 != 0
pckt.ACK = transLayer[13]&0x10 != 0
pckt.Lost = uint32(cp.Length - cp.CaptureLength)
pckt.Payload = ndata[dOf:]
return nil
pckt.Payload = copySlice(pckt.Payload, ndata[dOf:])
return
}
func (pckt *Packet) MessageID() uint64 {
@@ -232,12 +172,6 @@ func (pckt *Packet) Dst() string {
return fmt.Sprintf("%s:%d", pckt.DstIP, pckt.DstPort)
}
type EmptyPacket string
func (err EmptyPacket) Error() string {
return "Empty packet"
}
// ErrHdrLength returned on short header length
type ErrHdrLength string
+168 -96
View File
@@ -43,19 +43,14 @@ func generateHeader(request bool, seq uint32, length uint16) []byte {
func GetPackets(request bool, start uint32, _len int, payload []byte) []*Packet {
var packets = make([]*Packet, _len)
var err error
for i := start; i < start+uint32(_len); i++ {
d := append(generateHeader(request, i, uint16(len(payload))), payload...)
ci := &gopacket.CaptureInfo{Length: len(d), CaptureLength: len(d), Timestamp: time.Now()}
packets[i-start], err = ParsePacket(d, int(layers.LinkTypeLoop), 4, ci, true)
if request {
packets[i-start].Direction = DirIncoming
if len(payload) > 0 {
packets[i-start], _ = ParsePacket(d, int(layers.LinkTypeLoop), 4, ci)
} else {
packets[i-start].Direction = DirOutcoming
}
if err != nil {
panic(err)
packets[i-start] = new(Packet)
}
}
return packets
@@ -63,44 +58,50 @@ func GetPackets(request bool, start uint32, _len int, payload []byte) []*Packet
func TestRequestResponseMapping(t *testing.T) {
packets := []*Packet{
{SrcPort: 60000, DstPort: 80, Ack: 1, Seq: 1, Direction: DirIncoming, Timestamp: time.Unix(1, 0), Payload: []byte("GET / HTTP/1.1\r\n")},
{SrcPort: 60000, DstPort: 80, Ack: 1, Seq: 17, Direction: DirIncoming, Timestamp: time.Unix(2, 0), Payload: []byte("Host: localhost\r\n\r\n")},
{SrcPort: 60000, DstPort: 80, Ack: 1, Seq: 1, Timestamp: time.Unix(1, 0), Payload: []byte("GET / HTTP/1.1\r\n")},
{SrcPort: 60000, DstPort: 80, Ack: 1, Seq: 17, Timestamp: time.Unix(2, 0), Payload: []byte("Host: localhost\r\n\r\n")},
// Seq of first response packet match Ack of first request packet
{SrcPort: 80, DstPort: 60000, Ack: 36, Seq: 1, Direction: DirOutcoming, Timestamp: time.Unix(3, 0), Payload: []byte("HTTP/1.1 200 OK\r\n")},
{SrcPort: 80, DstPort: 60000, Ack: 36, Seq: 18, Direction: DirOutcoming, Timestamp: time.Unix(4, 0), Payload: []byte("Content-Length: 0\r\n\r\n")},
{SrcPort: 80, DstPort: 60000, Ack: 36, Seq: 1, Timestamp: time.Unix(3, 0), Payload: []byte("HTTP/1.1 200 OK\r\n")},
{SrcPort: 80, DstPort: 60000, Ack: 36, Seq: 18, Timestamp: time.Unix(4, 0), Payload: []byte("Content-Length: 0\r\n\r\n")},
// Same TCP stream
{SrcPort: 60000, DstPort: 80, Ack: 39, Seq: 36, Direction: DirIncoming, Timestamp: time.Unix(5, 0), Payload: []byte("GET / HTTP/1.1\r\n")},
{SrcPort: 60000, DstPort: 80, Ack: 39, Seq: 52, Direction: DirIncoming, Timestamp: time.Unix(6, 0), Payload: []byte("Host: localhost\r\n\r\n")},
{SrcPort: 60000, DstPort: 80, Ack: 39, Seq: 36, Timestamp: time.Unix(5, 0), Payload: []byte("GET / HTTP/1.1\r\n")},
{SrcPort: 60000, DstPort: 80, Ack: 39, Seq: 52, Timestamp: time.Unix(6, 0), Payload: []byte("Host: localhost\r\n\r\n")},
// Seq of first response packet match Ack of first request packet
{SrcPort: 80, DstPort: 60000, Ack: 71, Seq: 39, Direction: DirOutcoming, Timestamp: time.Unix(7, 0), Payload: []byte("HTTP/1.1 200 OK\r\n")},
{SrcPort: 80, DstPort: 60000, Ack: 71, Seq: 56, Direction: DirOutcoming, Timestamp: time.Unix(8, 0), Payload: []byte("Content-Length: 0\r\n\r\n")},
{SrcPort: 80, DstPort: 60000, Ack: 71, Seq: 39, Timestamp: time.Unix(7, 0), Payload: []byte("HTTP/1.1 200 OK\r\n")},
{SrcPort: 80, DstPort: 60000, Ack: 71, Seq: 56, Timestamp: time.Unix(8, 0), Payload: []byte("Content-Length: 0\r\n\r\n")},
}
parser := NewMessageParser(nil, nil, nil, time.Second, false)
var mssg = make(chan *Message, 4)
parser := NewMessageParser(1<<20, time.Second, nil, func(m *Message) { mssg <- m })
parser.Start = func(pckt *Packet) (bool, bool) {
return proto.HasRequestTitle(pckt.Payload), proto.HasResponseTitle(pckt.Payload)
}
parser.End = func(m *Message) bool {
return proto.HasFullPayload(m, m.PacketData()...)
return proto.HasFullPayload(m, m.Data())
}
for _, packet := range packets {
parser.processPacket(packet)
parser.PacketHandler(packet)
}
messages := []*Message{}
for i := 0; i < 4; i++ {
m := parser.Read()
messages = append(messages, m)
select {
case <-time.After(time.Second):
t.Errorf("can't parse packets fast enough")
return
case m := <-mssg:
messages = append(messages, m)
}
}
assert.Equal(t, int(messages[0].Direction), int(DirIncoming))
assert.Equal(t, int(messages[1].Direction), int(DirOutcoming))
assert.Equal(t, int(messages[2].Direction), int(DirIncoming))
assert.Equal(t, int(messages[3].Direction), int(DirOutcoming))
assert.Equal(t, messages[0].IsRequest, true)
assert.Equal(t, messages[1].IsRequest, false)
assert.Equal(t, messages[2].IsRequest, true)
assert.Equal(t, messages[3].IsRequest, false)
assert.Equal(t, messages[0].UUID(), messages[1].UUID())
assert.Equal(t, messages[2].UUID(), messages[3].UUID())
@@ -109,35 +110,36 @@ func TestRequestResponseMapping(t *testing.T) {
}
func TestMessageParserWithHint(t *testing.T) {
parser := NewMessageParser(nil, nil, nil, time.Second, false)
var mssg = make(chan *Message, 3)
parser := NewMessageParser(1<<20, time.Second, nil, func(m *Message) { mssg <- m })
parser.Start = func(pckt *Packet) (bool, bool) {
return proto.HasRequestTitle(pckt.Payload), proto.HasResponseTitle(pckt.Payload)
}
parser.End = func(m *Message) bool {
return proto.HasFullPayload(m, m.PacketData()...)
return proto.HasFullPayload(m, m.Data())
}
packets := GetPackets(true, 1, 30, nil)
packets[4] = GetPackets(false, 4, 1, []byte("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nTransfer-Encoding: chunked\r\n\r\n7"))[0]
packets[5] = GetPackets(false, 5, 1, []byte("\r\nMozilla\r\n9\r\nDeveloper\r"))[0]
packets[6] = GetPackets(false, 6, 1, []byte("\n7\r\nNetwork\r\n0\r\n\r\n"))[0]
packets[14] = GetPackets(true, 14, 1, []byte("POST / HTTP/1.1\r\nContent-Type: text/plain\r\nContent-Length: 23\r\n\r\n"))[0]
packets[15] = GetPackets(true, 15, 1, []byte("MozillaDeveloper"))[0]
packets[16] = GetPackets(true, 16, 1, []byte("Network"))[0]
packets[24] = GetPackets(true, 24, 1, []byte("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: 0\r\n\r\n"))[0]
packets := []*Packet{
// Seq of first response packet match Ack of first request packet
{SrcPort: 80, DstPort: 60000, Ack: 1, Seq: 1, Direction: DirOutcoming, Timestamp: time.Unix(1, 0), Payload: []byte("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nTransfer-Encoding: chunked\r\n\r\n7\r\n")},
{SrcPort: 80, DstPort: 60000, Ack: 1, Seq: 18, Direction: DirOutcoming, Timestamp: time.Unix(2, 0), Payload: []byte("\r\nMozilla\r\n9\r\nDeveloper\r")},
{SrcPort: 80, DstPort: 60000, Ack: 1, Seq: 42, Direction: DirOutcoming, Timestamp: time.Unix(3, 0), Payload: []byte("\n7\r\nNetwork\r\n0\r\n\r\n")},
{SrcPort: 60000, DstPort: 80, Ack: 60, Seq: 1, Direction: DirIncoming, Timestamp: time.Unix(4, 0), Payload: []byte("POST / HTTP/1.1\r\nContent-Type: text/plain\r\nContent-Length: 23\r\n\r\n")},
{SrcPort: 60000, DstPort: 80, Ack: 60, Seq: 66, Direction: DirIncoming, Timestamp: time.Unix(5, 0), Payload: []byte("MozillaDeveloper")},
{SrcPort: 60000, DstPort: 80, Ack: 60, Seq: 82, Direction: DirIncoming, Timestamp: time.Unix(6, 0), Payload: []byte("Network")},
{SrcPort: 80, DstPort: 60000, Ack: 89, Seq: 1, Direction: DirOutcoming, Timestamp: time.Unix(7, 0), Payload: []byte("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: 0\r\n\r\n")},
}
for _, p := range packets {
parser.processPacket(p)
for i := 0; i < 30; i++ {
parser.PacketHandler(packets[i])
}
messages := []*Message{}
for i := 0; i < 3; i++ {
m := parser.Read()
messages = append(messages, m)
select {
case <-time.After(time.Second):
t.Errorf("can't parse packets fast enough")
return
case m := <-mssg:
messages = append(messages, m)
}
}
if !bytes.HasSuffix(messages[0].Data(), []byte("\n7\r\nNetwork\r\n0\r\n\r\n")) {
@@ -154,92 +156,158 @@ func TestMessageParserWithHint(t *testing.T) {
}
func TestMessageParserWrongOrder(t *testing.T) {
parser := NewMessageParser(nil, nil, nil, time.Second, false)
var mssg = make(chan *Message, 3)
parser := NewMessageParser(1<<20, time.Second, nil, func(m *Message) { mssg <- m })
parser.Start = func(pckt *Packet) (bool, bool) {
return proto.HasRequestTitle(pckt.Payload), proto.HasResponseTitle(pckt.Payload)
}
parser.End = func(m *Message) bool {
return proto.HasFullPayload(m, m.PacketData()...)
return proto.HasFullPayload(m, m.Data())
}
packets := []*Packet{
// Seq of first response packet match Ack of first request packet
{SrcPort: 60000, DstPort: 80, Ack: 60, Seq: 66, Direction: DirIncoming, Timestamp: time.Unix(5, 0), Payload: []byte("MozillaDeveloper")},
{SrcPort: 80, DstPort: 60000, Ack: 1, Seq: 1, Direction: DirOutcoming, Timestamp: time.Unix(1, 0), Payload: []byte("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nTransfer-Encoding: chunked\r\n\r\n7\r\n")},
{SrcPort: 80, DstPort: 60000, Ack: 1, Seq: 42, Direction: DirOutcoming, Timestamp: time.Unix(3, 0), Payload: []byte("\n7\r\nNetwork\r\n0\r\n\r\n")},
packets := GetPackets(true, 1, 30, nil)
packets[6] = GetPackets(false, 4, 1, []byte("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nTransfer-Encoding: chunked\r\n\r\n7"))[0]
packets[5] = GetPackets(false, 5, 1, []byte("\r\nMozilla\r\n9\r\nDeveloper\r"))[0]
packets[4] = GetPackets(false, 6, 1, []byte("\n7\r\nNetwork\r\n0\r\n\r\n"))[0]
// Duplicate with same seq
packets[7] = GetPackets(false, 6, 1, []byte("\n7\r\nNetwork\r\n0\r\n\r\n"))[0]
{SrcPort: 60000, DstPort: 80, Ack: 60, Seq: 1, Direction: DirIncoming, Timestamp: time.Unix(4, 0), Payload: []byte("POST / HTTP/1.1\r\nContent-Type: text/plain\r\nContent-Length: 23\r\n\r\n")},
{SrcPort: 80, DstPort: 60000, Ack: 1, Seq: 18, Direction: DirOutcoming, Timestamp: time.Unix(2, 0), Payload: []byte("\r\nMozilla\r\n9\r\nDeveloper\r")},
packets[16] = GetPackets(true, 14, 1, []byte("POST / HTTP/1.1\r\nContent-Type: text/plain\r\nContent-Length: 23\r\n\r\n"))[0]
packets[15] = GetPackets(true, 15, 1, []byte("MozillaDeveloper"))[0]
packets[14] = GetPackets(true, 16, 1, []byte("Network"))[0]
{SrcPort: 80, DstPort: 60000, Ack: 89, Seq: 1, Direction: DirOutcoming, Timestamp: time.Unix(7, 0), Payload: []byte("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: 0\r\n\r\n")},
{SrcPort: 60000, DstPort: 80, Ack: 60, Seq: 82, Direction: DirIncoming, Timestamp: time.Unix(6, 0), Payload: []byte("Network")},
for i := 0; i < 30; i++ {
parser.PacketHandler(packets[i])
}
for _, p := range packets {
parser.processPacket(p)
var m *Message
select {
case <-time.After(time.Second):
t.Errorf("can't parse packets fast enough")
return
case m = <-mssg:
}
m := parser.Read()
if !bytes.HasSuffix(m.Data(), []byte("\n7\r\nNetwork\r\n0\r\n\r\n")) {
t.Errorf("expected to %q to have suffix %q", m.Data(), []byte("\n7\r\nNetwork\r\n0\r\n\r\n"))
}
m = parser.Read()
if !bytes.HasSuffix(m.Data(), []byte("Content-Length: 0\r\n\r\n")) {
t.Errorf("expected to %q to have suffix %q", m.Data(), []byte("Content-Length: 0\r\n\r\n"))
select {
case <-time.After(time.Second):
t.Errorf("can't parse packets fast enough")
return
case m = <-mssg:
}
m = parser.Read()
if !bytes.HasSuffix(m.Data(), []byte("Network")) {
t.Errorf("expected to %q to have suffix %q", m.Data(), []byte("Network"))
}
}
func TestMessageParserWithoutHint(t *testing.T) {
var mssg = make(chan *Message, 1)
var data [63 << 10]byte
packets := GetPackets(true, 1, 10, data[:])
p := NewMessageParser(nil, nil, nil, time.Second, false)
p := NewMessageParser(63<<10*10, time.Second, nil, func(m *Message) { mssg <- m })
for _, v := range packets {
p.processPacket(v)
p.PacketHandler(v)
}
var m *Message
select {
case <-time.After(time.Second):
t.Errorf("can't parse packets fast enough")
return
case m = <-mssg:
}
m := p.Read()
if m.Length != 63<<10*10 {
t.Errorf("expected %d to equal %d", m.Length, 63<<10*10)
}
}
func TestMessageTimeoutReached(t *testing.T) {
const size = 63 << 11
var data [size >> 1]byte
func TestMessageMaxSizeReached(t *testing.T) {
var mssg = make(chan *Message, 2)
var data [63 << 10]byte
packets := GetPackets(true, 1, 2, data[:])
p := NewMessageParser(nil, nil, nil, 10*time.Millisecond, true)
p.processPacket(packets[0])
time.Sleep(time.Millisecond * 200)
p.processPacket(packets[1])
m := p.Read()
if m.Length != size {
t.Errorf("expected %d to equal %d", m.Length, size)
packets = append(packets, GetPackets(true, 1, 1, make([]byte, 63<<10+10))...)
p := NewMessageParser(63<<10+10, time.Second, nil, func(m *Message) { mssg <- m })
for _, v := range packets {
p.PacketHandler(v)
}
var m *Message
select {
case <-time.After(time.Second):
t.Errorf("can't parse packets fast enough")
return
case m = <-mssg:
}
if m.Length != 63<<10+10 {
t.Errorf("expected %d to equal %d", m.Length, 63<<10+10)
}
if !m.Truncated {
t.Error("expected message to be truncated")
}
select {
case <-time.After(time.Second):
t.Errorf("can't parse packets fast enough")
return
case m = <-mssg:
}
if m.Length != 63<<10+10 {
t.Errorf("expected %d to equal %d", m.Length, 63<<10+10)
}
if m.Truncated {
t.Error("expected message to not be truncated")
}
}
func TestMessageTimeoutReached(t *testing.T) {
var mssg = make(chan *Message, 2)
var data [63 << 10]byte
packets := GetPackets(true, 1, 2, data[:])
p := NewMessageParser(1<<20, 0, nil, func(m *Message) { mssg <- m })
p.PacketHandler(packets[0])
time.Sleep(time.Millisecond * 400)
p.PacketHandler(packets[1])
m := <-mssg
if m.Length != 63<<10 {
t.Errorf("expected %d to equal %d", m.Length, 63<<10)
}
if !m.TimedOut {
t.Error("expected message to be timeout")
}
}
func TestMessageUUID(t *testing.T) {
packets := GetPackets(true, 1, 10, nil)
var uuid, uuid1 []byte
parser := NewMessageParser(0, 0, nil, func(msg *Message) {
if len(uuid) == 0 {
uuid = msg.UUID()
return
}
uuid1 = msg.UUID()
})
for _, p := range packets {
parser.PacketHandler(p)
}
if string(uuid) != string(uuid1) {
t.Errorf("expected %s, to equal %s", uuid, uuid1)
}
}
func BenchmarkMessageUUID(b *testing.B) {
packets := GetPackets(true, 1, 5, nil)
var uuid []byte
parser := NewMessageParser(nil, nil, nil, 10*time.Millisecond, true)
var msg *Message
parser := NewMessageParser(0, 0, nil, func(m *Message) {
msg = m
})
for _, p := range packets {
parser.processPacket(p)
parser.PacketHandler(p)
}
msg := parser.Read()
b.ResetTimer()
for i := 0; i < b.N; i++ {
uuid = msg.UUID()
@@ -260,16 +328,20 @@ func BenchmarkPacketParseAndSort(b *testing.B) {
}
func BenchmarkMessageParserWithoutHint(b *testing.B) {
// runtime.GOMAXPROCS(8)
var mssg = make(chan *Message, 1)
var chunk = []byte("111111111111111111111111111111")
packets := GetPackets(true, 1, 1000, chunk)
p := NewMessageParser(nil, nil, nil, 2*time.Second, false)
p := NewMessageParser(1<<20, time.Second*2, nil, func(m *Message) {
mssg <- m
})
b.ResetTimer()
b.ReportMetric(float64(1000), "packets/op")
for i := 0; i < b.N; i++ {
for _, v := range packets {
p.processPacket(v)
p.PacketHandler(v)
}
p.Read()
<-mssg
}
}
@@ -285,22 +357,22 @@ func BenchmarkMessageParserWithHint(b *testing.B) {
for i := 0; i < len(buf); i++ {
packets[i] = GetPackets(false, 1, 1, buf[i])[0]
}
parser := NewMessageParser(nil, nil, nil, 2*time.Second, false)
var mssg = make(chan *Message, 1)
parser := NewMessageParser(1<<30, time.Second*10, nil, func(m *Message) { mssg <- m })
parser.Start = func(pckt *Packet) (bool, bool) {
return false, proto.HasResponseTitle(pckt.Payload)
}
parser.End = func(m *Message) bool {
return proto.HasFullPayload(m, m.PacketData()...)
return proto.HasFullPayload(m, m.Data())
}
b.ResetTimer()
b.ReportMetric(float64(len(packets)), "packets/op")
b.ReportMetric(float64(1000), "chunks/op")
for i := 0; i < b.N; i++ {
for j := range packets {
parser.processPacket(packets[j])
parser.PacketHandler(packets[j])
}
parser.Read()
<-mssg
}
}
@@ -308,6 +380,6 @@ func BenchmarkNewAndParsePacket(b *testing.B) {
data := append(generateHeader(true, 1024, 10), make([]byte, 10)...)
b.ResetTimer()
for i := 0; i < b.N; i++ {
ParsePacket(data, int(layers.LinkTypeLoop), 4, &gopacket.CaptureInfo{}, true)
ParsePacket(data, int(layers.LinkTypeLoop), 4, &gopacket.CaptureInfo{})
}
}