add id in raw message stats

emitter.go

@@ -9,12 +9,21 @@ import (
 	"time"
 
 	"github.com/buger/goreplay/byteutils"
+	"github.com/buger/goreplay/stat"
 )
 
+// EmitterConfig configurations associated with emitters
+type EmitterConfig struct {
+	Stats                bool
+	SplitOutput          bool
+	RecognizeTCPSessions bool
+}
+
 // Emitter represents an abject to manage plugins communication
 type Emitter struct {
 	sync.WaitGroup
 	plugins *InOutPlugins
+	st      *stat.Writer
 }
 
 // NewEmitter creates and initializes new Emitter object.
@@ -28,6 +37,10 @@ func (e *Emitter) Start(plugins *InOutPlugins, middlewareCmd string) {
 		Settings.CopyBufferSize = 5 << 20
 	}
 	e.plugins = plugins
+	if Settings.EmitterConfig.Stats {
+		e.st = initializeStatWriter("meta.csv", "type: GOR_STAT\ntitle: Messages Meta")
+		statsWriteRecord(e.st, []string{"type", "id", "timestamp", "latency"}, "[EMITTER]")
+	}
 
 	if middlewareCmd != "" {
 		middleware := NewMiddleware(middlewareCmd)
@@ -41,8 +54,8 @@ func (e *Emitter) Start(plugins *InOutPlugins, middlewareCmd string) {
 		e.Add(1)
 		go func() {
 			defer e.Done()
-			if err := CopyMulty(middleware, plugins.Outputs...); err != nil {
-				Debug(2, fmt.Sprintf("[EMITTER] error during copy: %q", err))
+			if err := e.CopyMulty(middleware, plugins.Outputs...); err != nil {
+				Debug(1, fmt.Sprintf("[EMITTER] error during copy: %q", err))
 			}
 		}()
 	} else {
@@ -50,8 +63,8 @@ func (e *Emitter) Start(plugins *InOutPlugins, middlewareCmd string) {
 			e.Add(1)
 			go func(in PluginReader) {
 				defer e.Done()
-				if err := CopyMulty(in, plugins.Outputs...); err != nil {
-					Debug(2, fmt.Sprintf("[EMITTER] error during copy: %q", err))
+				if err := e.CopyMulty(in, plugins.Outputs...); err != nil {
+					Debug(1, fmt.Sprintf("[EMITTER] error during copy: %q", err))
 				}
 			}(in)
 		}
@@ -70,10 +83,13 @@ func (e *Emitter) Close() {
 		e.Wait()
 	}
 	e.plugins.All = nil // avoid Close to make changes again
+	if e.st != nil {
+		e.st.Close()
+	}
 }
 
 // CopyMulty copies from 1 reader to multiple writers
-func CopyMulty(src PluginReader, writers ...PluginWriter) error {
+func (e *Emitter) CopyMulty(src PluginReader, writers ...PluginWriter) error {
 	wIndex := 0
 	modifier := NewHTTPModifier(&Settings.ModifierConfig)
 	filteredRequests := make(map[string]int64)
@@ -97,13 +113,16 @@ func CopyMulty(src PluginReader, writers ...PluginWriter) error {
 				Debug(2, fmt.Sprintf("[EMITTER] Found malformed record %q from %q", msg.Meta, src))
 				continue
 			}
+			if e.st != nil {
+				statsWriteRecord(e.st, splitUnsafe(meta), "[EMITTER]")
+			}
 			requestID := byteutils.SliceToString(meta[1])
 			// start a subroutine only when necessary
 			if Settings.Verbose >= 3 {
 				Debug(3, "[EMITTER] input: ", byteutils.SliceToString(msg.Meta[:len(msg.Meta)-1]), " from: ", src)
 			}
 			if modifier != nil {
-				Debug(3, "[EMITTER] modifier:", requestID, "from:", src)
+				Debug(3, "[EMITTER] modifier: ", requestID, " from: ", src)
 				if isRequestPayload(msg.Meta) {
 					msg.Data = modifier.Rewrite(msg.Data)
 					// If modifier tells to skip request
@@ -175,3 +194,11 @@ func CopyMulty(src PluginReader, writers ...PluginWriter) error {
 		}
 	}
 }
+
+func splitUnsafe(data [][]byte) []string {
+	s := make([]string, len(data), len(data))
+	for i := range data {
+		s[i] = byteutils.SliceToString(data[i])
+	}
+	return s
+}

gor_stat.go

@@ -1,64 +0,0 @@
-package main
-
-import (
-	"runtime"
-	"strconv"
-	"time"
-)
-
-type GorStat struct {
-	statName string
-	rateMs   int
-	latest   int
-	mean     int
-	max      int
-	count    int
-}
-
-func NewGorStat(statName string, rateMs int) (s *GorStat) {
-	s = new(GorStat)
-	s.statName = statName
-	s.rateMs = rateMs
-	s.latest = 0
-	s.mean = 0
-	s.max = 0
-	s.count = 0
-
-	if Settings.Stats {
-		go s.reportStats()
-	}
-	return
-}
-
-func (s *GorStat) Write(latest int) {
-	if Settings.Stats {
-		if latest > s.max {
-			s.max = latest
-		}
-		if latest != 0 {
-			s.mean = ((s.mean * s.count) + latest) / (s.count + 1)
-		}
-		s.latest = latest
-		s.count = s.count + 1
-	}
-}
-
-func (s *GorStat) Reset() {
-	s.latest = 0
-	s.max = 0
-	s.mean = 0
-	s.count = 0
-}
-
-func (s *GorStat) String() string {
-	return s.statName + ":" + strconv.Itoa(s.latest) + "," + strconv.Itoa(s.mean) + "," + strconv.Itoa(s.max) + "," + strconv.Itoa(s.count) + "," + strconv.Itoa(s.count/(s.rateMs/1000.0)) + "," + strconv.Itoa(runtime.NumGoroutine())
-}
-
-func (s *GorStat) reportStats() {
-	Debug(0, "\n", s.statName+":latest,mean,max,count,count/second,gcount")
-	for {
-		Debug(0, "\n", s)
-		s.Reset()
-		time.Sleep(time.Duration(s.rateMs) * time.Millisecond)
-	}
-}

input_raw.go

@@ -12,6 +12,7 @@ import (
 	"github.com/buger/goreplay/capture"
 	"github.com/buger/goreplay/proto"
 	"github.com/buger/goreplay/size"
+	"github.com/buger/goreplay/stat"
 	"github.com/buger/goreplay/tcp"
 )
 
@@ -68,7 +69,7 @@ type RAWInputConfig struct {
 type RAWInput struct {
 	sync.Mutex
 	RAWInputConfig
-	messageStats   []tcp.Stats
+	st             *stat.Writer
 	listener       *capture.Listener
 	message        chan *tcp.Message
 	cancelListener context.CancelFunc
@@ -96,6 +97,13 @@ func NewRAWInput(address string, config RAWInputConfig) (i *RAWInput) {
 	}
 	i.host = host
 	i.port = uint16(port)
+	if i.Stats {
+		i.st = initializeStatWriter("input_raw_message.csv", "type: GOR_STAT\ntitle: raw messages info")
+		statsWriteRecord(i.st, []string{
+			"length", "lostdata", "source", "destination", "ipversion",
+			"timedout", "truncated", "id",
+		}, "[INPUT-RAW]")
+	}
 
 	i.listen(address)
 
@@ -108,30 +116,28 @@ func (i *RAWInput) PluginRead() (*Message, error) {
 	var msg Message
 	select {
 	case <-i.quit:
+		if i.st != nil {
+			i.st.Close()
+			i.st = nil
+		}
 		return nil, ErrorStopped
 	case msgTCP = <-i.message:
 		msg.Data = msgTCP.Data()
 	}
 	var msgType byte = ResponsePayload
+	timestamp := msgTCP.Start.UnixNano()
+	latency := msgTCP.End.UnixNano() - timestamp
 	if msgTCP.IsIncoming {
 		msgType = RequestPayload
 		if i.RealIPHeader != "" {
 			msg.Data = proto.SetHeader(msg.Data, []byte(i.RealIPHeader), []byte(msgTCP.SrcAddr))
 		}
 	}
-	msg.Meta = payloadHeader(msgType, msgTCP.UUID(), msgTCP.Start.UnixNano(), msgTCP.End.UnixNano()-msgTCP.Start.UnixNano())
+	id := msgTCP.UUID()
+	msg.Meta = payloadHeader(msgType, id, timestamp, latency)
 
-	// to be removed....
-	if msgTCP.Truncated {
-		Debug(2, "[INPUT-RAW] message truncated, increase copy-buffer-size")
-	}
-	// to be removed...
-	if msgTCP.TimedOut {
-		Debug(2, "[INPUT-RAW] message timeout reached, increase input-raw-expire")
-	}
-	if i.Stats {
-		stat := msgTCP.Stats
-		go i.addStats(stat)
+	if i.Stats && i.st != nil {
+		statsWriteRecord(i.st, getStats(msgTCP.Stats, id), "[INPUT-RAW]")
 	}
 	msgTCP = nil
 	return &msg, nil
@@ -173,13 +179,26 @@ func (i *RAWInput) String() string {
 }
 
 // GetStats returns the stats so far and reset the stats
-func (i *RAWInput) GetStats() []tcp.Stats {
-	i.Lock()
-	defer func() {
-		i.messageStats = []tcp.Stats{}
-		i.Unlock()
-	}()
-	return i.messageStats
+func getStats(stat tcp.Stats, id []byte) []string {
+	var s [8]string
+
+	s[0] = strconv.Itoa(stat.Length)    // length
+	s[1] = strconv.Itoa(stat.LostData)  // lostdata
+	s[2] = stat.SrcAddr                 // source
+	s[3] = stat.DstAddr                 // destination
+	s[4] = string(stat.IPversion + '0') // ipversion
+	if stat.TimedOut {                  // timedout
+		s[5] = "true"
+	} else {
+		s[5] = "false"
+	}
+	if stat.Truncated { // truncated
+		s[6] = "true"
+	} else {
+		s[6] = "false"
+	}
+	s[7] = string(id)
+	return s[:]
 }
 
 // Close closes the input raw listener
@@ -189,15 +208,6 @@ func (i *RAWInput) Close() error {
 	return nil
 }
 
-func (i *RAWInput) addStats(mStats tcp.Stats) {
-	i.Lock()
-	if len(i.messageStats) >= 10000 {
-		i.messageStats = []tcp.Stats{}
-	}
-	i.messageStats = append(i.messageStats, mStats)
-	i.Unlock()
-}
-
 func http1StartHint(pckt *tcp.Packet) (isIncoming, isOutgoing bool) {
 	isIncoming = proto.HasRequestTitle(pckt.Payload)
 	if isIncoming {

output_binary.go

@@ -30,7 +30,6 @@ type BinaryOutput struct {
 	needWorker    chan int
 	quit          chan struct{}
 	config        *BinaryOutputConfig
-	queueStats    *GorStat
 }
 
 // NewBinaryOutput constructor for BinaryOutput

output_http.go

@@ -32,12 +32,10 @@ type response struct {
 // HTTPOutputConfig struct for holding http output configuration
 type HTTPOutputConfig struct {
 	TrackResponses bool          `json:"output-http-track-response"`
-	Stats          bool          `json:"output-http-stats"`
 	OriginalHost   bool          `json:"output-http-original-host"`
 	RedirectLimit  int           `json:"output-http-redirect-limit"`
 	WorkersMin     int           `json:"output-http-workers-min"`
 	WorkersMax     int           `json:"output-http-workers"`
-	StatsMs        int           `json:"output-http-stats-ms"`
 	QueueLen       int           `json:"output-http-queue-len"`
 	ElasticSearch  string        `json:"output-http-elasticsearch"`
 	Timeout        time.Duration `json:"output-http-timeout"`
@@ -54,7 +52,6 @@ type HTTPOutputConfig struct {
 type HTTPOutput struct {
 	activeWorkers int32
 	config        *HTTPOutputConfig
-	queueStats    *GorStat
 	elasticSearch *ESPlugin
 	client        *HTTPClient
 	stopWorker    chan struct{}
@@ -105,9 +102,6 @@ func NewHTTPOutput(address string, config *HTTPOutputConfig) PluginReadWriter {
 	}
 	o.config = config
 	o.stop = make(chan bool)
-	if o.config.Stats {
-		o.queueStats = NewGorStat("output_http", o.config.StatsMs)
-	}
 
 	o.queue = make(chan *Message, o.config.QueueLen)
 	if o.config.TrackResponses {
@@ -179,9 +173,6 @@ func (o *HTTPOutput) PluginWrite(msg *Message) (n int, err error) {
 	case o.queue <- msg:
 	}
 
-	if o.config.Stats {
-		o.queueStats.Write(len(o.queue))
-	}
 	if len(o.queue) > 0 {
 		// try to start a new worker to serve
 		if atomic.LoadInt32(&o.activeWorkers) < int32(o.config.WorkersMax) {

output_tcp.go

@@ -15,9 +15,9 @@ type TCPOutput struct {
 	address     string
 	limit       int
 	buf         []chan *Message
-	bufStats    *GorStat
 	config      *TCPOutputConfig
 	workerIndex uint32
+	stop        chan struct{}
 }
 
 // TCPOutputConfig tcp output configuration
@@ -35,10 +35,7 @@ func NewTCPOutput(address string, config *TCPOutputConfig) PluginWriter {
 
 	o.address = address
 	o.config = config
-
-	if Settings.OutputTCPStats {
-		o.bufStats = NewGorStat("output_tcp", 5000)
-	}
+	o.stop = make(chan struct{})
 
 	// create X buffers and send the buffer index to the worker
 	o.buf = make([]chan *Message, o.config.Workers)
@@ -72,7 +69,12 @@ func (o *TCPOutput) worker(bufferIndex int) {
 	defer conn.Close()
 
 	for {
-		msg := <-o.buf[bufferIndex]
+		var msg *Message
+		select {
+		case <-o.stop:
+			return
+		case msg = <-o.buf[bufferIndex]:
+		}
 		if _, err = conn.Write(msg.Meta); err == nil {
 			if _, err = conn.Write(msg.Data); err == nil {
 				_, err = conn.Write(payloadSeparatorAsBytes)
@@ -109,10 +111,6 @@ func (o *TCPOutput) PluginWrite(msg *Message) (n int, err error) {
 	bufferIndex := o.getBufferIndex(msg.Data)
 	o.buf[bufferIndex] <- msg
 
-	if Settings.OutputTCPStats {
-		o.bufStats.Write(len(o.buf[bufferIndex]))
-	}
-
 	return len(msg.Data) + len(msg.Meta), nil
 }
 
@@ -129,3 +127,9 @@ func (o *TCPOutput) connect(address string) (conn net.Conn, err error) {
 func (o *TCPOutput) String() string {
 	return fmt.Sprintf("TCP output %s, limit: %d", o.address, o.limit)
 }
+
+// Close closes the data channel so that data
+func (o *TCPOutput) Close() error {
+	close(o.stop)
+	return nil
+}

settings.go

@@ -3,7 +3,10 @@ package main
 import (
 	"flag"
 	"fmt"
+	"log"
 	"os"
+	"path/filepath"
+	"strings"
 	"sync"
 	"time"
 )
@@ -27,12 +30,10 @@ func (h *MultiOption) Set(value string) error {
 // AppSettings is the struct of main configuration
 type AppSettings struct {
 	Verbose   int           `json:"verbose"`
-	Stats     bool          `json:"stats"`
 	ExitAfter time.Duration `json:"exit-after"`
 
-	SplitOutput          bool   `json:"split-output"`
-	RecognizeTCPSessions bool   `json:"recognize-tcp-sessions"`
-	Pprof                string `json:"http-pprof"`
+	EmitterConfig
+	Pprof string `json:"http-pprof"`
 
 	InputDummy   MultiOption `json:"input-dummy"`
 	OutputDummy  MultiOption
@@ -43,7 +44,6 @@ type AppSettings struct {
 	InputTCPConfig  TCPInputConfig
 	OutputTCP       MultiOption `json:"output-tcp"`
 	OutputTCPConfig TCPOutputConfig
-	OutputTCPStats  bool `json:"output-tcp-stats"`
 
 	InputFile        MultiOption `json:"input-file"`
 	InputFileLoop    bool        `json:"input-file-loop"`
@@ -80,11 +80,10 @@ func usage() {
 	os.Exit(2)
 }
 
-func init() {
+func initFlags() {
 	flag.Usage = usage
 	flag.StringVar(&Settings.Pprof, "http-pprof", "", "Enable profiling. Starts  http server on specified port, exposing special /debug/pprof endpoint. Example: `:8181`")
 	flag.IntVar(&Settings.Verbose, "verbose", 0, "set the level of verbosity, if greater than zero then it will turn on debug output")
-	flag.BoolVar(&Settings.Stats, "stats", false, "Turn on queue stats output")
 
 	if DEMO == "" {
 		flag.DurationVar(&Settings.ExitAfter, "exit-after", 0, "exit after specified duration")
@@ -92,6 +91,7 @@ func init() {
 		Settings.ExitAfter = 5 * time.Minute
 	}
 
+	flag.BoolVar(&Settings.EmitterConfig.Stats, "meta-stats", false, "Turn on stats for meta data.")
 	flag.BoolVar(&Settings.SplitOutput, "split-output", false, "By default each output gets same traffic. If set to `true` it splits traffic equally among all outputs.")
 	flag.BoolVar(&Settings.RecognizeTCPSessions, "recognize-tcp-sessions", false, "[PRO] If turned on http output will create separate worker for each TCP session. Splitting output will session based as well.")
 
@@ -109,7 +109,6 @@ func init() {
 	flag.BoolVar(&Settings.OutputTCPConfig.SkipVerify, "output-tcp-skip-verify", false, "Don't verify hostname on TLS secure connection.")
 	flag.BoolVar(&Settings.OutputTCPConfig.Sticky, "output-tcp-sticky", false, "Use Sticky connection. Request/Response with same ID will be sent to the same connection.")
 	flag.IntVar(&Settings.OutputTCPConfig.Workers, "output-tcp-workers", 10, "Number of parallel tcp connections, default is 10")
-	flag.BoolVar(&Settings.OutputTCPStats, "output-tcp-stats", false, "Report TCP output queue stats to console every 5 seconds.")
 
 	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.")
@@ -140,7 +139,7 @@ func init() {
 	flag.Var(&Settings.BufferSize, "input-raw-buffer-size", "Controls size of the OS buffer which holds packets until they dispatched. Default value depends by system: in Linux around 2MB. If you see big package drop, increase this value.")
 	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.RAWInputConfig.Stats, "input-raw-stats", false, "enable stats generator on raw TCP messages")
 
 	flag.StringVar(&Settings.Middleware, "middleware", "", "Used for modifying traffic using external command")
 
@@ -153,13 +152,9 @@ func init() {
 	flag.IntVar(&Settings.OutputHTTPConfig.QueueLen, "output-http-queue-len", 1000, "Number of requests that can be queued for output, if all workers are busy. default = 1000")
 	flag.BoolVar(&Settings.OutputHTTPConfig.SkipVerify, "output-http-skip-verify", false, "Don't verify hostname on TLS secure connection.")
 	flag.DurationVar(&Settings.OutputHTTPConfig.WorkerTimeout, "output-http-worker-timeout", 2*time.Second, "Duration to rollback idle workers.")
-
 	flag.IntVar(&Settings.OutputHTTPConfig.RedirectLimit, "output-http-redirects", 0, "Enable how often redirects should be followed.")
 	flag.DurationVar(&Settings.OutputHTTPConfig.Timeout, "output-http-timeout", 5*time.Second, "Specify HTTP request/response timeout. By default 5s. Example: --output-http-timeout 30s")
 	flag.BoolVar(&Settings.OutputHTTPConfig.TrackResponses, "output-http-track-response", false, "If turned on, HTTP output responses will be set to all outputs like stdout, file and etc.")
-
-	flag.BoolVar(&Settings.OutputHTTPConfig.Stats, "output-http-stats", false, "Report http output queue stats to console every N milliseconds. See output-http-stats-ms")
-	flag.IntVar(&Settings.OutputHTTPConfig.StatsMs, "output-http-stats-ms", 5000, "Report http output queue stats to console every N milliseconds. default: 5000")
 	flag.BoolVar(&Settings.OutputHTTPConfig.OriginalHost, "http-original-host", false, "Normally gor replaces the Host http header with the host supplied with --output-http.  This option disables that behavior, preserving the original Host header.")
 	flag.StringVar(&Settings.OutputHTTPConfig.ElasticSearch, "output-http-elasticsearch", "", "Send request and response stats to ElasticSearch:\n\tgor --input-raw :8080 --output-http staging.com --output-http-elasticsearch 'es_host:api_port/index_name'")
 	/* outputHTTPConfig */
@@ -234,3 +229,32 @@ func Debug(level int, args ...interface{}) {
 		fmt.Fprintln(os.Stderr, args...)
 	}
 }
+
+func initGorDir() {
+	var err error
+	p := os.Getenv("GORDIR")
+	if p == "" {
+		if p, err = os.UserHomeDir(); err != nil {
+			goto errLog
+		}
+		if !strings.Contains(p, "root") {
+			p = filepath.Join(p, ".gor")
+		} else {
+			p = ".gor"
+		}
+	}
+	if err = os.MkdirAll(p, os.ModeDir|os.ModePerm); err != nil {
+		goto errLog
+	}
+	if err = os.Setenv("GORDIR", p); err != nil {
+		goto errLog
+	}
+	return
+errLog:
+	log.Fatalf("init GORDIR error:%q", err)
+}
+
+func init() {
+	initGorDir()
+	initFlags()
+}

stat/csv/writer.go

@@ -0,0 +1,47 @@
+package csv
+
+import (
+	"encoding/csv"
+	"os"
+	"strings"
+)
+
+// Writer check encoding/csv.Writer
+type Writer struct {
+	*csv.Writer
+	f *os.File
+}
+
+// NewWriter returns an csv writer with file opened
+// error returned will be associated with opening the file
+func NewWriter(path string) (*Writer, error) {
+	var wr Writer
+	f, err := os.Create(path)
+	if err != nil {
+		return nil, err
+	}
+	wr.f = f
+	wr.Writer = csv.NewWriter(f)
+	return &wr, nil
+}
+
+// WriteHeader will write infos about this docs in form of comments
+// it uses "#" to indicate that this should be ignored by the reader
+func (wr *Writer) WriteHeader(hdr string) error {
+	_, err := wr.f.WriteString(formatHeader(hdr))
+	return err
+}
+
+// Close the underlying file
+func (wr *Writer) Close() error {
+	return wr.f.Close()
+}
+
+func formatHeader(hdr string) string {
+	if !strings.HasSuffix(hdr, "\n") {
+		hdr += "\n"
+	}
+	// add comments before every line
+	hdr = strings.ReplaceAll("#"+hdr, "\n", "\n#")
+	return hdr[:len(hdr)-1]
+}

stat/csv/writer_test.go

@@ -0,0 +1,23 @@
+package csv
+
+import (
+	"testing"
+)
+
+func TestFormatHeader(t *testing.T) {
+	m := map[string]string{
+		"hello":                       "#hello\n",
+		"":                            "#\n",
+		"hello\n":                     "#hello\n",
+		"hello\nhello\n":              "#hello\n#hello\n",
+		"hello\nhello":                "#hello\n#hello\n",
+		"hello\nhello\nhello":         "#hello\n#hello\n#hello\n",
+		"hello\nhello\nhello\n":       "#hello\n#hello\n#hello\n",
+		"hello\nhello\r\n\r\nhello\n": "#hello\n#hello\r\n#\r\n#hello\n",
+	}
+	for k, v := range m {
+		if vl := formatHeader(k); vl != v {
+			t.Errorf("expected %q to equal %q", vl, v)
+		}
+	}
+}

stat/writer.go

@@ -0,0 +1,124 @@
+package stat
+
+import (
+	"errors"
+	"sync"
+	"time"
+)
+
+// ErrNotImplemented returned when impls requested are not implemnted
+var ErrNotImplemented = errors.New("method not implemented")
+
+// RecordWriter represents a stat writer it should have all this method.
+// for rationale  behind this interface check encoding/csv or
+type RecordWriter interface {
+	Write(record []string) error
+	Flush()
+	Error() error
+}
+
+const flushSpan = 10
+
+// Writer represents a single instance of goreplay stats writer
+type Writer struct {
+	name    string
+	records [flushSpan][]string // a temporary stats holder
+	off     int
+	quit    chan struct{}
+	mu      sync.Mutex
+	w       RecordWriter
+	err     error
+}
+
+// NewWriter returns a new instance of stats associated with the writer "wr"
+// and "name" as the stats identity
+func NewWriter(w RecordWriter, name string) (wr *Writer) {
+	wr = new(Writer)
+	wr.name = name
+	wr.w = w
+	wr.quit = make(chan struct{})
+	return
+}
+
+// WriteHeader writes header of the documents to the underlying writer.
+// Header can be any kind of data like json unmarshal or file info.
+// underlying writer should sets a different way to represents the header.
+func (wr *Writer) WriteHeader(header string) error {
+	if wHdr, ok := wr.w.(interface{ WriteHeader(header string) error }); ok {
+		return wHdr.WriteHeader(header)
+	}
+	return ErrNotImplemented
+}
+
+// SetReader sets another source of records reader to use along with (*Writer).Write([]string)
+// rd function will be called every "rate" duration time
+func (wr *Writer) SetReader(rd func() []string, rate time.Duration) {
+	go func() {
+		timer := time.NewTicker(rate)
+		defer timer.Stop()
+		for {
+			select {
+			case <-wr.quit:
+				return
+			case <-timer.C:
+				err := wr.Write(rd())
+				if err != nil {
+					return
+				}
+			}
+		}
+	}()
+}
+
+func (wr *Writer) Write(record []string) error {
+	wr.mu.Lock()
+	defer wr.mu.Unlock()
+	if wr.err != nil {
+		return wr.err
+	}
+	if len(record) > 0 {
+		wr.records[wr.off] = record
+		wr.off++
+	}
+	return wr.flush(wr.off >= flushSpan)
+}
+
+func (wr *Writer) String() string {
+	return "stats: " + wr.name
+}
+
+// Error reports any error that has occurred during a previous Write or Flush.
+func (wr *Writer) Error() error {
+	wr.mu.Lock()
+	wr.mu.Unlock()
+	return wr.err
+}
+
+// Close closes and flush all data of the stat
+func (wr *Writer) Close() error {
+	wr.mu.Lock()
+	defer wr.mu.Unlock()
+	close(wr.quit)
+	wr.err = wr.flush(true)
+	if cl, ok := wr.w.(interface{ Close() error }); ok {
+		wr.err = cl.Close()
+	}
+	return wr.err
+}
+
+func (wr *Writer) flush(f bool) error {
+	if !f {
+		return nil
+	}
+	i := 0
+	for i < wr.off {
+		if wr.err = wr.w.Write(wr.records[i]); wr.err != nil {
+			break
+		}
+		i++
+	}
+	wr.off -= i
+	wr.w.Flush()
+	wr.err = wr.w.Error()
+	return wr.err
+}

stat/writer_test.go

@@ -0,0 +1,33 @@
+package stat
+
+import "testing"
+
+type testWriter struct {
+	f  func([]string)
+	fl func()
+}
+
+func (t testWriter) WriteHeader(_ string) error { return nil }
+
+func (t testWriter) Flush() { t.fl() }
+
+func (t testWriter) Error() error { return nil }
+
+func (t testWriter) Write(r []string) error {
+	t.f(r)
+	return nil
+}
+
+func TestTempSize(t *testing.T) {
+	count := 0
+	tW := testWriter{f: func(_ []string) { count++ }, fl: func() {
+		if count != flushSpan {
+			t.Errorf("should flush after %d records", flushSpan)
+		}
+	}}
+	s := NewWriter(tW, "")
+	defer s.Close()
+	for i := 0; i < flushSpan; i++ {
+		s.Write([]string{"1"})
+	}
+}

stats.go

@@ -0,0 +1,35 @@
+package main
+
+import (
+	"log"
+	"os"
+	"path/filepath"
+
+	"github.com/buger/goreplay/stat"
+	"github.com/buger/goreplay/stat/csv"
+)
+
+func statsWriteRecord(st *stat.Writer, record []string, plugin string) {
+	if st == nil {
+		return
+	}
+	if st.Error() != nil {
+		Debug(1, plugin, " stats closed with error: ", st.Error())
+		st.Close()
+		st = nil
+		return
+	}
+	st.Write(record)
+}
+
+func initializeStatWriter(file string, hdr string) *stat.Writer {
+	csvWr, err := csv.NewWriter(filepath.Join(os.Getenv("GORDIR"), file))
+	if err != nil {
+		log.Fatalf("init meta stat writer error: %q", err)
+	}
+	st := stat.NewWriter(csvWr, "meta")
+	if hdr != "" {
+		st.WriteHeader("type: GOR_STATS\ntitle: message meta")
+	}
+	return st
+}

tcp/tcp_message.go

@@ -7,7 +7,6 @@ import (
 	"sort"
 	"sync"
 	"time"
-	"unsafe"
 
 	"github.com/buger/goreplay/size"
 	"github.com/google/gopacket"
@@ -55,10 +54,10 @@ func (m *Message) UUID() []byte {
 		lst := len(pckt.SrcIP) - 4
 		if m.IsIncoming {
 			key = uint64(pckt.SrcPort)<<48 | uint64(pckt.DstPort)<<32 |
-				uint64(_uint32(&pckt.SrcIP[lst]))
+				uint64(_uint32(pckt.SrcIP[lst:]))
 		} else {
 			key = uint64(pckt.DstPort)<<48 | uint64(pckt.SrcPort)<<32 |
-				uint64(_uint32(&pckt.DstIP[lst]))
+				uint64(_uint32(pckt.DstIP[lst:]))
 		}
 		if uuidHex, ok := m.pool.uuids[key]; ok {
 			delete(m.pool.uuids, key)
@@ -69,9 +68,7 @@ func (m *Message) UUID() []byte {
 	id := make([]byte, 12, 12)
 	binary.BigEndian.PutUint32(id, pckt.Seq)
 	tStamp := m.End.UnixNano()
-	for i, v := range _8byte(&tStamp) {
-		id[i+4] = v
-	}
+	binary.BigEndian.PutUint64(id[4:], uint64(tStamp))
 	uuidHex := make([]byte, 24, 24)
 	hex.Encode(uuidHex[:], id[:])
 	if m.pool.uuids != nil {
@@ -175,7 +172,7 @@ func (pool *MessagePool) Handler(packet gopacket.Packet) {
 	defer pool.Unlock()
 	lst := len(pckt.SrcIP) - 4
 	key := uint64(pckt.SrcPort)<<48 | uint64(pckt.DstPort)<<32 |
-		uint64(_uint32(&pckt.SrcIP[lst]))
+		uint64(_uint32(pckt.SrcIP[lst:]))
 	m, ok := pool.pool[key]
 	if pckt.RST {
 		if ok {
@@ -183,7 +180,7 @@ func (pool *MessagePool) Handler(packet gopacket.Packet) {
 			<-m.done
 		}
 		key = uint64(pckt.DstPort)<<48 | uint64(pckt.SrcPort)<<32 |
-			uint64(_uint32(&pckt.DstIP[lst]))
+			uint64(_uint32(pckt.DstIP[lst:]))
 		m, ok = pool.pool[key]
 		if ok {
 			m.done <- true
@@ -230,8 +227,8 @@ func (pool *MessagePool) cleanUUIDs() {
 	now := time.Now().UnixNano()
 	for k, v := range pool.uuids {
 		// there is a timestamp wrapped in every ID
-		tStamp = _int64(&v[4])
-		if time.Duration(now-tStamp) >= pool.messageExpire {
+		tStamp = int64(binary.BigEndian.Uint64(v[4:]))
+		if time.Duration(now-tStamp) > pool.messageExpire {
 			delete(pool.uuids, k)
 		}
 	}
@@ -289,16 +286,6 @@ func (pool *MessagePool) say(level int, args ...interface{}) {
 	}
 }
 
-type void = unsafe.Pointer
-
-func _uint32(b *byte) uint32 {
-	return *(*uint32)(void(b))
-}
-
-func _int64(b *byte) int64 {
-	return *(*int64)(void(b))
-}
-
-func _8byte(i *int64) [8]byte {
-	return *(*[8]byte)(void(i))
+func _uint32(b []byte) uint32 {
+	return binary.BigEndian.Uint32(b)
 }