Add missing PRO functionality and improve license protection

Author: buger

.gitignore

@@ -11,9 +11,11 @@ vendor
 hey
 
 *.bin
-
+lib/
+output/
 *.gz
 *.zip
+.aider*
 
 *.class
 
@@ -32,4 +34,4 @@ goreplay
 corpus
 crashers
 suppressions
-dist
+dist

LICENSE.txt

@@ -3,7 +3,7 @@ Copyright (c) 2011-present Leonid Bugaev
 Portions of this software are licensed as follows:
 
 * All content residing under the "doc/" directory of this repository is licensed under "Creative Commons: CC BY-SA 4.0 license".
-* Files with the following names "pro.go", "s3_reader.go", "output_s3.go" are released under the commercial license specified in "COMM-LICENSE" file.
+* The file "pro.go" and all files ending with the "_pro.go" suffix are released under the commercial license specified in the "COMM-LICENSE" file.
 * Content outside of the above mentioned directories or restrictions above is available under the "LGPLv3" license as defined below.
 
 

Makefile

@@ -158,4 +158,4 @@ bash:
 	$(RUN) /bin/bash
 
 dist:
-	mkdir -p $(DIST_PATH)
+	mkdir -p $(DIST_PATH)

ce.go

@@ -0,0 +1,17 @@
+//go:build !pro
+
+package goreplay
+
+import (
+	"fmt"
+)
+
+// PRO this value indicates if goreplay is running in PRO mode.
+var PRO = false
+
+func SettingsHook(settings *AppSettings) {
+	if settings.RecognizeTCPSessions {
+		settings.RecognizeTCPSessions = false
+		fmt.Println("[ERROR] TCP session recognition is not supported in the open-source version of GoReplay")
+	}
+}

go.mod

@@ -1,6 +1,6 @@
 module github.com/buger/goreplay
 
-go 1.19
+go 1.21
 
 require (
 	github.com/Shopify/sarama v1.38.1

go.sum

@@ -5,6 +5,7 @@ github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAE
 github.com/Shopify/sarama v1.38.1 h1:lqqPUPQZ7zPqYlWpTh+LQ9bhYNu2xJL6k1SJN4WVe2A=
 github.com/Shopify/sarama v1.38.1/go.mod h1:iwv9a67Ha8VNa+TifujYoWGxWnu2kNVAQdSdZ4X2o5g=
 github.com/Shopify/toxiproxy/v2 v2.5.0 h1:i4LPT+qrSlKNtQf5QliVjdP08GyAH8+BUIc9gT0eahc=
+github.com/Shopify/toxiproxy/v2 v2.5.0/go.mod h1:yhM2epWtAmel9CB8r2+L+PCmhH6yH2pITaPAo7jxJl0=
 github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
 github.com/araddon/gou v0.0.0-20211019181548-e7d08105776c h1:XUqw//RExYoxW4Eie8MuKp8sEDAZI1gMHX/daUFgZww=
 github.com/araddon/gou v0.0.0-20211019181548-e7d08105776c/go.mod h1:ikc1XA58M+Rx7SEbf0bLJCfBkwayZ8T5jBo5FXK8Uz8=
@@ -47,6 +48,7 @@ github.com/envoyproxy/go-control-plane v0.9.9-0.20210512163311-63b5d3c536b0/go.m
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
 github.com/flowstack/go-jsonschema v0.1.1/go.mod h1:yL7fNggx1o8rm9RlgXv7hTBWxdBM0rVwpMwimd3F3N0=
 github.com/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8vw=
+github.com/fortytw2/leaktest v1.3.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
 github.com/go-logr/logr v1.2.0/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
 github.com/go-logr/logr v1.2.4 h1:g01GSCwiDw2xSZfjJ2/T9M+S6pFdcNtFYsp+Y43HYDQ=
@@ -58,6 +60,7 @@ github.com/go-openapi/jsonreference v0.20.2/go.mod h1:Bl1zwGIM8/wsvqjsOQLJ/SH+En
 github.com/go-openapi/swag v0.22.3 h1:yMBqmnQ0gyZvEb/+KzuWZOXgllrXT4SADYbvDaXHv/g=
 github.com/go-openapi/swag v0.22.3/go.mod h1:UzaqsxGiab7freDnrUUra0MwWfN/q7tE4j+VcZ0yl14=
 github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0 h1:p104kn46Q8WdvHunIJ9dAyjPVtrBPhSr3KT2yUst43I=
+github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0/go.mod h1:fyg7847qk6SyHyPtNmDHnmrv/HOrqktSC+C9fM+CJOE=
 github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
 github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
@@ -96,6 +99,7 @@ github.com/google/gofuzz v1.2.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/
 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/pprof v0.0.0-20210720184732-4bb14d4b1be1 h1:K6RDEckDVWvDI9JAJYCmNdQXq6neHJOYx3V6jnqNEec=
+github.com/google/pprof v0.0.0-20210720184732-4bb14d4b1be1/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
 github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
 github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
@@ -143,6 +147,7 @@ github.com/klauspost/compress v1.16.5/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQs
 github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
 github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
 github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0=
+github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
@@ -159,7 +164,9 @@ github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjY
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
 github.com/onsi/ginkgo/v2 v2.9.1 h1:zie5Ly042PD3bsCvsSOPvRnFwyo3rKe64TJlD6nu0mk=
+github.com/onsi/ginkgo/v2 v2.9.1/go.mod h1:FEcmzVcCHl+4o9bQZVab+4dC9+j+91t2FHSzmGAPfuo=
 github.com/onsi/gomega v1.27.4 h1:Z2AnStgsdSayCMDiCU42qIz+HLqEPcgiOCXjAU/w+8E=
+github.com/onsi/gomega v1.27.4/go.mod h1:riYq/GJKh8hhoM01HN6Vmuy93AarCXCBGpvFDK3q3fQ=
 github.com/pierrec/lz4/v4 v4.1.17 h1:kV4Ip+/hUBC+8T6+2EgburRtkE9ef4nbY3f4dFhGjMc=
 github.com/pierrec/lz4/v4 v4.1.17/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@@ -170,12 +177,14 @@ github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 h1:N/ElC8H3+5X
 github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
 github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
 github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
+github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
 github.com/smartystreets/assertions v1.2.0 h1:42S6lae5dvLc7BrLu/0ugRtcFVjoJNMC/N3yZFZkDFs=
 github.com/smartystreets/assertions v1.2.0/go.mod h1:tcbTF8ujkAEcZ8TElKY+i30BzYlVhC/LOxJk7iOWnoo=
 github.com/smartystreets/goconvey v1.7.2 h1:9RBaZCeXEQ3UselpuwUQHltGVXvdwm6cv1hgR6gDIPg=
 github.com/smartystreets/goconvey v1.7.2/go.mod h1:Vw0tHAZW6lzCRk3xgdin6fKYcG+G3Pg9vgXWeJpQFMM=
 github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
 github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
+github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
 github.com/stoewer/go-strcase v1.2.0/go.mod h1:IBiWB2sKIp3wVVQ3Y035++gc+knqhUQag1KpM8ahLw8=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
@@ -253,6 +262,7 @@ golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o=
+golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -302,6 +312,7 @@ golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roY
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
 golang.org/x/tools v0.7.0 h1:W4OVu8VVOaIO0yzWMNdepAulS7YfoS3Zabrm8DOXXU4=
+golang.org/x/tools v0.7.0/go.mod h1:4pg6aUX35JBAogB10C9AtvVL+qowtN4pT3CGSQex14s=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

input_file.go

@@ -238,6 +238,16 @@ func NewFileInput(path string, loop bool, readDepth int, maxWait time.Duration,
 	return
 }
 
+func parseS3Url(path string) (bucket, key string) {
+	path = path[5:] // stripping `s3://`
+	sep := strings.IndexByte(path, '/')
+
+	bucket = path[:sep]
+	key = path[sep+1:]
+
+	return bucket, key
+}
+
 func (i *FileInput) init() (err error) {
 	defer i.mu.Unlock()
 	i.mu.Lock()

output_binary.go

@@ -1,11 +1,16 @@
+//go:build !pro
+
 package goreplay
 
 import (
-	"github.com/buger/goreplay/internal/size"
-	"sync/atomic"
+	"errors"
 	"time"
+
+	"github.com/buger/goreplay/internal/size"
 )
 
+var _ PluginWriter = (*BinaryOutput)(nil)
+
 // BinaryOutputConfig struct for holding binary output configuration
 type BinaryOutputConfig struct {
 	Workers        int           `json:"output-binary-workers"`
@@ -19,155 +24,30 @@ type BinaryOutputConfig struct {
 // By default workers pool is dynamic and starts with 10 workers
 // You can specify fixed number of workers using `--output-tcp-workers`
 type BinaryOutput struct {
-	// Keep this as first element of struct because it guarantees 64bit
-	// alignment. atomic.* functions crash on 32bit machines if operand is not
-	// aligned at 64bit. See https://github.com/golang/go/issues/599
-	activeWorkers int64
-	address       string
-	queue         chan *Message
-	responses     chan response
-	needWorker    chan int
-	quit          chan struct{}
-	config        *BinaryOutputConfig
-	queueStats    *GorStat
+	address string
 }
 
 // NewBinaryOutput constructor for BinaryOutput
 // Initialize workers
 func NewBinaryOutput(address string, config *BinaryOutputConfig) PluginReadWriter {
-	o := new(BinaryOutput)
-
-	o.address = address
-	o.config = config
-
-	o.queue = make(chan *Message, 1000)
-	o.responses = make(chan response, 1000)
-	o.needWorker = make(chan int, 1)
-	o.quit = make(chan struct{})
-
-	// Initial workers count
-	if o.config.Workers == 0 {
-		o.needWorker <- initialDynamicWorkers
-	} else {
-		o.needWorker <- o.config.Workers
-	}
-
-	go o.workerMaster()
-
-	return o
-}
-
-func (o *BinaryOutput) workerMaster() {
-	for {
-		newWorkers := <-o.needWorker
-		for i := 0; i < newWorkers; i++ {
-			go o.startWorker()
-		}
-
-		// Disable dynamic scaling if workers poll fixed size
-		if o.config.Workers != 0 {
-			return
-		}
-	}
+	return &BinaryOutput{address: address}
 }
 
-func (o *BinaryOutput) startWorker() {
-	client := NewTCPClient(o.address, &TCPClientConfig{
-		Debug:              o.config.Debug,
-		Timeout:            o.config.Timeout,
-		ResponseBufferSize: int(o.config.BufferSize),
-	})
-
-	deathCount := 0
-
-	atomic.AddInt64(&o.activeWorkers, 1)
-
-	for {
-		select {
-		case msg := <-o.queue:
-			o.sendRequest(client, msg)
-			deathCount = 0
-		case <-time.After(time.Millisecond * 100):
-			// When dynamic scaling enabled workers die after 2s of inactivity
-			if o.config.Workers == 0 {
-				deathCount++
-			} else {
-				continue
-			}
-
-			if deathCount > 20 {
-				workersCount := atomic.LoadInt64(&o.activeWorkers)
-
-				// At least 1 startWorker should be alive
-				if workersCount != 1 {
-					atomic.AddInt64(&o.activeWorkers, -1)
-					return
-				}
-			}
-		}
-	}
-}
-
-// PluginWrite writes a message tothis plugin
+// PluginWrite writes a message to this plugin
 func (o *BinaryOutput) PluginWrite(msg *Message) (n int, err error) {
-	if !isRequestPayload(msg.Meta) {
-		return len(msg.Data), nil
-	}
-
-	o.queue <- msg
-
-	if o.config.Workers == 0 {
-		workersCount := atomic.LoadInt64(&o.activeWorkers)
-
-		if len(o.queue) > int(workersCount) {
-			o.needWorker <- len(o.queue)
-		}
-	}
-
-	return len(msg.Data) + len(msg.Meta), nil
+	return 0, errors.New("binary output is only available in PRO version")
 }
 
 // PluginRead reads a message from this plugin
 func (o *BinaryOutput) PluginRead() (*Message, error) {
-	var resp response
-	var msg Message
-	select {
-	case <-o.quit:
-		return nil, ErrorStopped
-	case resp = <-o.responses:
-	}
-	msg.Data = resp.payload
-	msg.Meta = payloadHeader(ReplayedResponsePayload, resp.uuid, resp.startedAt, resp.roundTripTime)
-
-	return &msg, nil
-}
-
-func (o *BinaryOutput) sendRequest(client *TCPClient, msg *Message) {
-	if !isRequestPayload(msg.Meta) {
-		return
-	}
-
-	uuid := payloadID(msg.Meta)
-
-	start := time.Now()
-	resp, err := client.Send(msg.Data)
-	stop := time.Now()
-
-	if err != nil {
-		Debug(1, "Request error:", err)
-	}
-
-	if o.config.TrackResponses {
-		o.responses <- response{resp, uuid, start.UnixNano(), stop.UnixNano() - start.UnixNano()}
-	}
+	return nil, errors.New("binary output is only available in PRO version")
 }
 
 func (o *BinaryOutput) String() string {
-	return "Binary output: " + o.address
+	return "Binary output: " + o.address + " (PRO version required)"
 }
 
 // Close closes this plugin for reading
 func (o *BinaryOutput) Close() error {
-	close(o.quit)
 	return nil
 }