Create probe.yaml

Adding a new flag `--input-kafka-offset` for supporting consumes from Kafka by specified offset.

.github/workflows/probe.yaml

@@ -0,0 +1,38 @@
+name: AI Comment Handler
+
+
+on:
+  pull_request:
+    types: [opened] #[opened , labeled]
+  issue_comment:
+    types: [created]          
+  issues:
+    types: [opened] #[opened, labeled]  
+    
+# Define permissions needed for the workflow
+permissions:
+  issues: write
+  pull-requests: write
+  contents: read
+
+jobs:
+  trigger_probe_chat:
+    # Uncomment if you want to run on on specific lables, in this example `probe`
+    # if: |
+    #   (github.event_name == 'pull_request' && github.event.action == 'opened') || 
+    #   (github.event_name == 'issues' && github.event.action == 'opened') || 
+    #   (github.event_name == 'issue_comment' && github.event.action == 'created') || 
+    #   ((github.event_name == 'pull_request' || github.event_name == 'issues') && 
+    #    github.event.action == 'labeled' && github.event.label.name == 'probe')
+    # Use the reusable workflow from your repository (replace <your-org/repo> and <ref>)
+    uses: buger/probe/.github/workflows/probe.yml@main
+    # Pass required inputs
+    with:
+      command_prefix: "/probe" # Or '/ai', '/ask', etc.
+      # Optionally override the default npx command if the secret isn't set
+      # default_probe_chat_command: 'node path/to/custom/script.js'
+    # Pass ALL secrets from this repository to the reusable workflow
+    # This includes GITHUB_TOKEN, PROBE_CHAT_COMMAND (if set), ANTHROPIC_API_KEY, etc.
+    secrets:
+      ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
+      ANTHROPIC_API_URL: ${{ secrets.ANTHROPIC_API_URL }}

.gitignore

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

COMM-LICENSE

@@ -52,9 +52,7 @@ Notwithstanding the foregoing, Licensor represents and warrants that it either o
 
 9. Limitation of Liability. 
 
-In no event will either party be liable for any direct, indirect, consequential, incidental, special, exemplary, or punitive damages or liabilities whatsoever arising from or relating to the Software, the Software content or this Agreement, whether based on contract, tort (including negligence), strict liability or other theory, even if such party has been advised of the possibility of such damages.
-
-In no event will Licensor liability exceed the Software license price as indicated in the invoice. The existence of more than one claim will not enlarge or extend this limit.
+In no event will either party be liable for any direct, indirect, consequential, incidental, special, exemplary, or punitive damages or liabilities whatsoever arising from or relating to the Software, the Software content or this Agreement, whether based on contract, tort (including negligence), strict liability or other theory, even if such party has been advised of the possibility of such damages. Except for indemnification obligations in Section 13.5, in no event will Licensor liability exceed the Software license price as indicated in the invoice. The existence of more than one claim will not enlarge or extend this limit.
 
 10. Remedies. Your exclusive remedy and Licensor’ entire liability for breach of this Agreement shall be limited, at Licensor’ sole and exclusive discretion, to (a) replacement of any defective software or documentation; or (b) refund of the license fee paid to Licensor, payable in accordance with Licensor' refund policy.
 
@@ -79,7 +77,7 @@ In no event will Licensor liability exceed the Software license price as indicat
 
 13.4 Export Compliance. Each party agrees to comply with all applicable laws and regulations, including laws, regulations, orders or other restrictions on export, re-export or redistribution of software.
 
-13.5 Indemnification. You agree to defend, indemnify, and hold harmless Licensor from and against any lawsuits, claims, losses, damages, fines and expenses (including attorneys' fees and costs) arising out of your use of the Software or breach of this Agreement.
+13.5 Indemnification. Licensor agrees to defend, indemnify, and hold harmless You from and against any lawsuits, claims, losses, damages, fines and expenses (including attorneys' fees and costs) arising out of a claim brought by a third-party alleging that the Software directly infringes such third-party’s intellectual property rights; provided that Licensor shall have no indemnity obligation for claims arising out of your modification to the Software. You agree to defend, indemnify, and hold harmless Licensor from and against any lawsuits, claims, losses, damages, fines and expenses (including attorneys' fees and costs) brought by a third-party arising out of your modifications to the Software or breach of this Agreement.
 
 13.6 Governing Law. This Agreement is governed by the laws of the State of Oregon and the United States without regard to conflicts of laws provisions thereof, and without regard to the United Nations Convention on the International Sale of Goods or the Uniform Computer Information Transactions Act, as currently enacted by any jurisdiction or as may be codified or amended from time to time by any jurisdiction. The jurisdiction and venue for actions related to the subject matter hereof shall be the state of Oregon and United States federal courts located in Portland, Oregon, and both parties hereby submit to the personal jurisdiction of such courts. 
 

Dockerfile.dev

@@ -1,8 +1,16 @@
-FROM golang:1.19
+ARG BASE_IMAGE
+FROM ${BASE_IMAGE}
 
-RUN apt-get update && apt-get install ruby vim-common -y
+RUN apk add --no-cache \
+    gcc \
+    g++ \
+    make \
+    linux-headers \
+    bison \
+    flex \
+    git \
+    wget
 
-RUN apt-get install flex bison -y
 RUN wget http://www.tcpdump.org/release/libpcap-1.10.0.tar.gz && tar xzf libpcap-1.10.0.tar.gz && cd libpcap-1.10.0 && ./configure && make install
 
 WORKDIR /go/src/github.com/buger/goreplay/

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

@@ -11,18 +11,19 @@ BENCHMARK = BenchmarkRAWInput
 TEST = TestRawListenerBench
 BIN_NAME = gor
 VERSION := DEV-$(shell date +%s)
+CUSTOM_TAGS := --tags "ngo$(if $(CUSTOM_BUILD_TAGS), $(CUSTOM_BUILD_TAGS),)"
 LDFLAGS = -ldflags "-X main.VERSION=$(VERSION) -extldflags \"-static\" -X main.DEMO=$(DEMO)"
 MAC_LDFLAGS = -ldflags "-X main.VERSION=$(VERSION) -X main.DEMO=$(DEMO)"
 DOCKER_FPM_CMD := docker run --rm -t -v `pwd`:/src -w /src fleetdm/fpm
 
 FPM_COMMON= \
-    --name $(PROJECT_NAME) \
-    --description "GoReplay is an open-source network monitoring tool which can record your live traffic, and use it for shadowing, load testing, monitoring and detailed analysis." \
-    -v $(VERSION) \
-    --vendor "Leonid Bugaev" \
-    -m "<support@goreplay.org>" \
-    --url "https://goreplay.org" \
-    -s dir
+				--name $(PROJECT_NAME) \
+				--description "GoReplay is an open-source network monitoring tool which can record your live traffic, and use it for shadowing, load testing, monitoring and detailed analysis." \
+				-v $(VERSION) \
+				--vendor "Leonid Bugaev" \
+				-m "<support@goreplay.org>" \
+				--url "https://goreplay.org" \
+				-s dir
 
 release: clean release-linux-amd64 release-linux-arm64 release-mac-amd64 release-mac-arm64 release-windows
 
@@ -31,19 +32,19 @@ vendor:
 	go mod vendor
 
 release-bin-linux-amd64: vendor
-	docker run --platform linux/amd64 --rm -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=amd64 -i $(CONTAINER_AMD) go build -mod=vendor -o $(BIN_NAME) -tags netgo $(LDFLAGS) ./cmd/gor/
+	docker run --platform linux/amd64 --rm -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=amd64 -i $(CONTAINER_AMD) go build -mod=vendor -o $(BIN_NAME) $(CUSTOM_TAGS) $(LDFLAGS) ./cmd/gor/
 
 release-bin-linux-arm64: vendor
-	docker run --platform linux/arm64 --rm -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=arm64 -i $(CONTAINER_ARM) go build -mod=vendor -o $(BIN_NAME) -tags netgo $(LDFLAGS) ./cmd/gor/
+	docker run --platform linux/arm64 --rm -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=arm64 -i $(CONTAINER_ARM) go build -mod=vendor -o $(BIN_NAME) $(CUSTOM_TAGS) $(LDFLAGS) ./cmd/gor/
 
 release-bin-mac-amd64: vendor
-	GOOS=darwin go build -mod=vendor -o $(BIN_NAME) $(MAC_LDFLAGS) ./cmd/gor/
+	GOOS=darwin go build -mod=vendor -o $(BIN_NAME) $(CUSTOM_TAGS) $(MAC_LDFLAGS) ./cmd/gor/
 
 release-bin-mac-arm64: vendor
-	GOOS=darwin GOARCH=arm64 CGO_ENABLED=1 go build -mod=vendor -o $(BIN_NAME) $(MAC_LDFLAGS)
+	GOOS=darwin GOARCH=arm64 CGO_ENABLED=1 go build -mod=vendor -o $(BIN_NAME) $(CUSTOM_TAGS) $(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.19.2-main --build-cmd "make VERSION=$(VERSION) build" -p "windows/amd64" ./cmd/gor/
+	docker run -it --rm -v `pwd`:$(SOURCE_PATH) -w $(SOURCE_PATH) -e CGO_ENABLED=1 docker.elastic.co/beats-dev/golang-crossbuild:1.19.2-main --build-cmd "make VERSION=$(VERSION) CUSTOM_BUILD_TAGS=$(CUSTOM_BUILD_TAGS) build" -p "windows/amd64" ./cmd/gor/
 	mv $(BIN_NAME) "$(BIN_NAME).exe"
 
 release-linux-amd64: dist release-bin-linux-amd64
@@ -78,18 +79,18 @@ clean:
 	rm -rf $(DIST_PATH)
 
 build:
-	go build -mod=vendor -o $(BIN_NAME) $(LDFLAGS)
+	go build -mod=vendor -o $(BIN_NAME) $(CUSTOM_TAGS) $(LDFLAGS)
 
 install:
-	go install $(MAC_LDFLAGS)
+	go install $(CUSTOM_TAGS) $(MAC_LDFLAGS)
 
 build-env: build-amd64-env build-arm64-env
 
 build-amd64-env:
-	docker buildx build --platform linux/amd64 -t $(CONTAINER_AMD) -f Dockerfile.dev .
+	docker buildx build --build-arg BASE_IMAGE=golang:1.22-alpine --platform linux/amd64 -t $(CONTAINER_AMD) -f Dockerfile.dev --load .
 
 build-arm64-env:
-	docker buildx build --platform linux/arm64 -t $(CONTAINER_ARM) -f Dockerfile.dev .
+	docker buildx build --build-arg BASE_IMAGE=arm64v8/golang:1.22-alpine --platform linux/arm64 -t $(CONTAINER_ARM) -f Dockerfile.dev --load .
 
 build-docker:
 	docker build -t gor-dev -f Dockerfile .
@@ -104,13 +105,13 @@ race:
 	$(RUN) go test ./... $(ARGS) -v -race -timeout 15s
 
 test:
-	$(RUN) go test ./. -timeout 120s $(LDFLAGS) $(ARGS)  -v
+	$(RUN) go test ./. -timeout 120s $(CUSTOM_TAGS) $(LDFLAGS) $(ARGS)  -v
 
 test_all:
-	$(RUN) go test ./... -timeout 120s $(LDFLAGS) $(ARGS) -v
+	$(RUN) go test ./... -timeout 120s $(CUSTOM_TAGS) $(LDFLAGS) $(ARGS) -v
 
 testone:
-	$(RUN) go test ./. -timeout 60s $(LDFLAGS) -run $(TEST) $(ARGS) -v
+	$(RUN) go test ./. -timeout 60s $(CUSTOM_TAGS) $(LDFLAGS) -run $(TEST) $(ARGS) -v
 
 cover:
 	$(RUN) go test $(ARGS) -race -v -timeout 15s -coverprofile=coverage.out
@@ -123,39 +124,39 @@ vet:
 	$(RUN) go vet
 
 bench:
-	$(RUN) go test $(LDFLAGS) -v -run NOT_EXISTING -bench $(BENCHMARK) -benchtime 5s
+	$(RUN) go test $(CUSTOM_TAGS) $(LDFLAGS) -v -run NOT_EXISTING -bench $(BENCHMARK) -benchtime 5s
 
 profile_test:
-	$(RUN) go test $(LDFLAGS) -run $(TEST) ./capture/. $(ARGS) -memprofile mem.mprof -cpuprofile cpu.out
-	$(RUN) go test $(LDFLAGS) -run $(TEST) ./capture/. $(ARGS) -c
+	$(RUN) go test $(CUSTOM_TAGS) $(LDFLAGS) -run $(TEST) ./capture/. $(ARGS) -memprofile mem.mprof -cpuprofile cpu.out
+	$(RUN) go test $(CUSTOM_TAGS) $(LDFLAGS) -run $(TEST) ./capture/. $(ARGS) -c
 
 # Used mainly for debugging, because docker container do not have access to parent machine ports
 run:
-	$(RUN) go run $(LDFLAGS) $(SOURCE) --input-dummy=0 --output-http="http://localhost:9000" --input-raw-track-response --input-raw 127.0.0.1:9000 --verbose 0 --middleware "./examples/middleware/echo.sh" --output-file requests.gor
+	$(RUN) go run $(CUSTOM_TAGS) $(LDFLAGS) $(SOURCE) --input-dummy=0 --output-http="http://localhost:9000" --input-raw-track-response --input-raw 127.0.0.1:9000 --verbose 0 --middleware "./examples/middleware/echo.sh" --output-file requests.gor
 
 run-2:
-	$(RUN) go run $(LDFLAGS) $(SOURCE) --input-raw :8000 --input-raw-bpf-filter "dst port 8000" --output-stdout --output-http "http://localhost:8000" --input-dummy=0
+	$(RUN) go run $(CUSTOM_TAGS) $(LDFLAGS) $(SOURCE) --input-raw :8000 --input-raw-bpf-filter "dst port 8000" --output-stdout --output-http "http://localhost:8000" --input-dummy=0
 
 run-3:
-	sudo -E go run $(SOURCE) --input-tcp :27001 --output-stdout
+	sudo -E go run $(CUSTOM_TAGS) $(SOURCE) --input-tcp :27001 --output-stdout
 
 run-arg:
-	sudo -E go run $(SOURCE) $(ARGS)
+	sudo -E go run $(CUSTOM_TAGS) $(SOURCE) $(ARGS)
 
 file-server:
-	go run $(SOURCE) file-server $(FADDR)
+	go run $(CUSTOM_TAGS) $(SOURCE) file-server $(FADDR)
 
 readpcap:
-	go run $(SOURCE) --input-raw $(FILE) --input-raw-track-response --input-raw-engine pcap_file --output-stdout
+	go run $(CUSTOM_TAGS) $(SOURCE) --input-raw $(FILE) --input-raw-track-response --input-raw-engine pcap_file --output-stdout
 
 record:
-	$(RUN) go run $(SOURCE) --input-dummy=0 --output-file=requests.gor --verbose --debug
+	$(RUN) go run $(CUSTOM_TAGS) $(SOURCE) --input-dummy=0 --output-file=requests.gor --verbose --debug
 
 replay:
-	$(RUN) go run $(SOURCE) --input-file=requests.bin --output-tcp=:9000 --verbose -h
+	$(RUN) go run $(CUSTOM_TAGS) $(SOURCE) --input-file=requests.bin --output-tcp=:9000 --verbose -h
 
 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
@@ -12,8 +12,8 @@ require (
 	github.com/mattbaird/elastigo v0.0.0-20170123220020-2fe47fd29e4b
 	github.com/stretchr/testify v1.8.2
 	github.com/xdg-go/scram v1.1.2
-	golang.org/x/net v0.10.0
-	golang.org/x/sys v0.8.0
+	golang.org/x/net v0.34.0
+	golang.org/x/sys v0.29.0
 	k8s.io/apimachinery v0.27.1
 	k8s.io/client-go v0.27.1
 )
@@ -60,10 +60,10 @@ require (
 	github.com/smartystreets/goconvey v1.7.2 // indirect
 	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
 	github.com/xdg-go/stringprep v1.0.4 // indirect
-	golang.org/x/crypto v0.9.0 // indirect
+	golang.org/x/crypto v0.32.0 // indirect
 	golang.org/x/oauth2 v0.8.0 // indirect
-	golang.org/x/term v0.8.0 // indirect
-	golang.org/x/text v0.9.0 // indirect
+	golang.org/x/term v0.28.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
 	golang.org/x/time v0.3.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
 	google.golang.org/protobuf v1.30.0 // indirect

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=
@@ -209,8 +218,8 @@ golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8U
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.6.0/go.mod h1:OFC/31mSvZgRz0V1QTNCzfAI1aIRzbiufJtkMIlEp58=
-golang.org/x/crypto v0.9.0 h1:LF6fAI+IutBocDJ2OT0Q1g8plpYljMZ4+lty+dsqw3g=
-golang.org/x/crypto v0.9.0/go.mod h1:yrmDGqONDYtNj3tH8X9dzUun2m2lzPa9ngI6/RUPGR0=
+golang.org/x/crypto v0.32.0 h1:euUpcYgM8WcP71gNpTqQCn6rC2t6ULUPiOzfWaXVVfc=
+golang.org/x/crypto v0.32.0/go.mod h1:ZnnJkOaASj8g0AjIduWNlq2NRxL0PlBrbKVyZ6V/Ugc=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
 golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU=
@@ -239,8 +248,8 @@ golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug
 golang.org/x/net v0.1.0/go.mod h1:Cx3nUiGt4eDBEyega/BKRp+/AlGL8hYe7U9odMt2Cco=
 golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
-golang.org/x/net v0.10.0 h1:X2//UzNDwYmtCLn7To6G58Wr6f5ahEAQgKNzv9Y951M=
-golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
+golang.org/x/net v0.34.0 h1:Mb7Mrk043xzHgnRM88suvJFwzVrRfHEHJEl5/71CKw0=
+golang.org/x/net v0.34.0/go.mod h1:di0qlW3YNM5oh6GqDGQr92MyTozJPmybPK4Ev/Gm31k=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.8.0 h1:6dkIjl3j3LtZ/O3sTgZTMsLKSftL/B8Zgq4huOIIUu8=
@@ -252,7 +261,8 @@ golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 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.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=
+golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 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=
@@ -269,14 +279,14 @@ golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.8.0 h1:EBmGv8NaZBZTWvrbjNoL6HVt+IVy3QDQpJs7VRIw3tU=
-golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.29.0 h1:TPYlXGxvx1MGTn2GiZDhnjPA9wZzZeGKHHmKhHYvgaU=
+golang.org/x/sys v0.29.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/term v0.1.0/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
-golang.org/x/term v0.8.0 h1:n5xxQn2i3PC0yLAbjTpNT85q/Kgzcr2gIoX9OrJUols=
-golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo=
+golang.org/x/term v0.28.0 h1:/Ts8HFuMR2E6IP/jlo7QVLZHggjKQbhu/7H0LJFr3Gg=
+golang.org/x/term v0.28.0/go.mod h1:Sw/lC2IAUZ92udQNf3WodGtn4k/XoLyZoh8v/8uiwek=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
@@ -286,8 +296,8 @@ golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
 golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
-golang.org/x/text v0.9.0 h1:2sjJmO8cDvYveuX97RDLsxlyUxLl+GHoLxBiRdHllBE=
-golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
+golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
+golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
 golang.org/x/time v0.3.0 h1:rg5rLMjNzMS1RkNLzCG38eapWhnYLFYXDXj2gOlr8j4=
 golang.org/x/time v0.3.0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
@@ -301,7 +311,8 @@ golang.org/x/tools v0.0.0-20200130002326-2f3ba24bd6e7/go.mod h1:TB2adYChydJhpapK
 golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 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.21.1-0.20240508182429-e35e4ccd0d2d h1:vU5i/LfpvrRCpgM/VPfJLg5KjxD3E+hfT1SH+d9zLwg=
+golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d/go.mod h1:aiJjzUbINMkxbQROHiO6hDPo2LHcIPhhQsa9DLh0yGk=
 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()

input_kafka.go

@@ -3,7 +3,9 @@ package goreplay
 import (
 	"encoding/json"
 	"log"
+	"strconv"
 	"strings"
+	"time"
 
 	"github.com/Shopify/sarama"
 	"github.com/Shopify/sarama/mocks"
@@ -12,14 +14,24 @@ import (
 // KafkaInput is used for receiving Kafka messages and
 // transforming them into HTTP payloads.
 type KafkaInput struct {
-	config    *InputKafkaConfig
-	consumers []sarama.PartitionConsumer
-	messages  chan *sarama.ConsumerMessage
-	quit      chan struct{}
+	config      *InputKafkaConfig
+	consumers   []sarama.PartitionConsumer
+	messages    chan *sarama.ConsumerMessage
+	speedFactor float64
+	quit        chan struct{}
+	kafkaTimer  *kafkaTimer
+}
+
+func getOffsetOfPartitions(offsetCfg string) int64 {
+	offset, err := strconv.ParseInt(offsetCfg, 10, 64)
+	if err != nil || offset < -2 {
+		log.Fatalln("Failed to parse offset: "+offsetCfg, err)
+	}
+	return offset
 }
 
 // NewKafkaInput creates instance of kafka consumer client with TLS config
-func NewKafkaInput(_ string, config *InputKafkaConfig, tlsConfig *KafkaTLSConfig) *KafkaInput {
+func NewKafkaInput(offsetCfg string, config *InputKafkaConfig, tlsConfig *KafkaTLSConfig) *KafkaInput {
 	c := NewKafkaConfig(&config.SASLConfig, tlsConfig)
 
 	var con sarama.Consumer
@@ -41,14 +53,17 @@ func NewKafkaInput(_ string, config *InputKafkaConfig, tlsConfig *KafkaTLSConfig
 	}
 
 	i := &KafkaInput{
-		config:    config,
-		consumers: make([]sarama.PartitionConsumer, len(partitions)),
-		messages:  make(chan *sarama.ConsumerMessage, 256),
-		quit:      make(chan struct{}),
+		config:      config,
+		consumers:   make([]sarama.PartitionConsumer, len(partitions)),
+		messages:    make(chan *sarama.ConsumerMessage, 256),
+		speedFactor: 1,
+		quit:        make(chan struct{}),
+		kafkaTimer:  new(kafkaTimer),
 	}
+	i.config.Offset = offsetCfg
 
 	for index, partition := range partitions {
-		consumer, err := con.ConsumePartition(config.Topic, partition, sarama.OffsetNewest)
+		consumer, err := con.ConsumePartition(config.Topic, partition, getOffsetOfPartitions(offsetCfg))
 		if err != nil {
 			log.Fatalln("Failed to start Sarama(Kafka) partition consumer:", err)
 		}
@@ -86,12 +101,15 @@ func (i *KafkaInput) PluginRead() (*Message, error) {
 	case message = <-i.messages:
 	}
 
+	inputTs := ""
+
 	msg.Data = message.Value
 	if i.config.UseJSON {
 
 		var kafkaMessage KafkaMessage
 		json.Unmarshal(message.Value, &kafkaMessage)
 
+		inputTs = kafkaMessage.ReqTs
 		var err error
 		msg.Data, err = kafkaMessage.Dump()
 		if err != nil {
@@ -103,8 +121,11 @@ func (i *KafkaInput) PluginRead() (*Message, error) {
 	// does it have meta
 	if isOriginPayload(msg.Data) {
 		msg.Meta, msg.Data = payloadMetaWithBody(msg.Data)
+		inputTs = string(payloadMeta(msg.Meta)[2])
 	}
 
+	i.timeWait(inputTs)
+
 	return &msg, nil
 
 }
@@ -118,3 +139,45 @@ func (i *KafkaInput) Close() error {
 	close(i.quit)
 	return nil
 }
+
+func (i *KafkaInput) timeWait(curInputTs string) {
+	if i.config.Offset == "-1" || curInputTs == "" {
+		return
+	}
+
+	// implement for Kafka input showdown or speedup emitting
+	timer := i.kafkaTimer
+	curTs := time.Now().UnixNano()
+
+	curInput, err := strconv.ParseInt(curInputTs, 10, 64)
+
+	if timer.latestInputTs == 0 || timer.latestOutputTs == 0 {
+		timer.latestInputTs = curInput
+		timer.latestOutputTs = curTs
+		return
+	}
+
+	if err != nil {
+		log.Fatalln("Fatal to parse timestamp err: ", err)
+	}
+
+	diffTs := curInput - timer.latestInputTs
+	pastTs := curTs - timer.latestOutputTs
+
+	diff := diffTs - pastTs
+	if i.speedFactor != 1 {
+		diff = int64(float64(diff) / i.speedFactor)
+	}
+
+	if diff > 0 {
+		time.Sleep(time.Duration(diff))
+	}
+
+	timer.latestInputTs = curInput
+	timer.latestOutputTs = curTs
+}
+
+type kafkaTimer struct {
+	latestInputTs  int64
+	latestOutputTs int64
+}

input_kafka_test.go

@@ -16,7 +16,7 @@ func TestInputKafkaRAW(t *testing.T) {
 		map[string][]int32{"test": {0}},
 	)
 
-	input := NewKafkaInput("", &InputKafkaConfig{
+	input := NewKafkaInput("-1", &InputKafkaConfig{
 		consumer: consumer,
 		Topic:    "test",
 		UseJSON:  false,
@@ -42,7 +42,7 @@ func TestInputKafkaJSON(t *testing.T) {
 		map[string][]int32{"test": {0}},
 	)
 
-	input := NewKafkaInput("", &InputKafkaConfig{
+	input := NewKafkaInput("-1", &InputKafkaConfig{
 		consumer: consumer,
 		Topic:    "test",
 		UseJSON:  true,

k8s/README.md

@@ -77,17 +77,24 @@ Following format supported:
 apiVersion: apps/v1
 kind: DaemonSet
 metadata:
-  name: goreplay-daemon
+  name: goreplay-daemon
 spec:
-  template:
-    spec:
+  selector:
+    matchLabels:
+      app: goreplay
+  template:
+    metadata:
+      labels:
+        app: goreplay
+    spec:
       hostNetwork: true
       serviceAccountName: goreplay
-      containers:
-      - name: goreplay
-        image: buger/goreplay:2.0.0-rc2
-        command:
-          - "--input-raw k8s://deployments/nginx:80"
+      containers:
+      - name: goreplay
+        image: buger/goreplay:2.0.0-rc4
+        args:
+          - "--input-raw"
+          - "k8s://deployments/nginx:80"
           - "--output-stdout"
 ```
 
@@ -135,10 +142,31 @@ spec:
 ```
 
 
-## 7. Verify installation
+## 7. Verify installation, and debugging tips
 
 Find url for your service using `kubectl get svc` or `minikube service --url ngnix-service -n http`, and make a call to it.
 
 Get GoReplay logs, and check if it capture traffic of your service.
-`kubectl logs -n goreplay -l name=goreplay-daemon --all-containers`
+`kubectl logs -n goreplay -l app=goreplay --all-containers`
 
+Describe daemonset:
+`kubectl describe daemonset goreplay-daemon -n goreplay`
+
+Get GoReplay pod list:
+`kubectl get pods -n goreplay -l app=goreplay`
+
+Get logs for specific pod (take data from previous step):
+`kubectl logs goreplay-daemon-<replace> -n goreplay`
+
+Get related k8s events:
+`kubectl get events -n goreplay --field-selector involvedObject.name=goreplay-daemon-<replace>`
+
+## 8. Debuggin telemetry
+
+We provde a special script which will get all require logs, and help you find installation errors. 
+
+`curl -s https://raw.githubusercontent.com/buger/goreplay/refs/heads/master/k8s/collect_goreplay_telemetry.sh | bash`
+
+If you are using microk8s or similar, you can also specific prefix for your kubectl command like this:
+
+`curl -s https://raw.githubusercontent.com/buger/goreplay/refs/heads/master/k8s/collect_goreplay_telemetry.sh | bash -s -- "microk8s kubectl"`

k8s/collect_goreplay_telemetry.sh

@@ -0,0 +1,123 @@
+#!/usr/bin/env bash
+#
+# collect_goreplay_telemetry.sh
+#
+# Gathers telemetry from a GoReplay DaemonSet in the 'goreplay' namespace.
+# Works on macOS and Linux, assuming 'kubectl' (or compatible) is installed.
+#
+# Usage examples:
+#   ./collect_goreplay_telemetry.sh
+#   ./collect_goreplay_telemetry.sh "microk8s kubectl"
+
+set -euo pipefail
+
+########################################
+# Determine kubectl command
+########################################
+if [[ $# -gt 0 ]]; then
+  # If an argument was provided, use that as the kubectl command
+  KUBECTL="$*"
+else
+  # Default to 'kubectl'
+  KUBECTL="kubectl"
+fi
+
+########################################
+# Check that the base command exists
+########################################
+# For "microk8s kubectl", we only check "microk8s" in PATH. For "oc" we check "oc".
+BASE_CMD="${KUBECTL%% *}"  # everything before the first space
+if ! command -v "${BASE_CMD}" >/dev/null 2>&1; then
+  echo "ERROR: '${BASE_CMD}' not found in PATH. Please install or configure it first."
+  exit 1
+fi
+
+echo "Using kubectl command: $KUBECTL"
+echo
+
+########################################
+# Helper function to print and run commands
+########################################
+run_cmd() {
+  echo "Command: $*"
+  eval "$*"
+}
+
+########################################
+# 1. Print logs from ALL GoReplay pods
+########################################
+echo "=================================================="
+echo "1. Gathering logs from all goreplay pods (all containers)..."
+echo "=================================================="
+run_cmd "$KUBECTL logs -n goreplay -l app=goreplay --all-containers" || {
+  echo "WARNING: Failed to get logs from pods with label app=goreplay"
+}
+
+########################################
+# 2. Describe the GoReplay DaemonSet
+########################################
+echo
+echo "=================================================="
+echo "2. Describing DaemonSet goreplay-daemon..."
+echo "=================================================="
+run_cmd "$KUBECTL describe daemonset goreplay-daemon -n goreplay" || {
+  echo "WARNING: Failed to describe daemonset goreplay-daemon"
+}
+
+########################################
+# 3. Get list of GoReplay pods (full output)
+########################################
+echo
+echo "=================================================="
+echo "3. Listing goreplay pods (full output)..."
+echo "=================================================="
+
+# Print full output (no -o name here):
+run_cmd "$KUBECTL get pods -n goreplay -l app=goreplay"
+
+# Then retrieve just the names for further processing:
+echo
+echo "Getting goreplay pod names for telemetry collection..."
+pods=$($KUBECTL get pods -n goreplay -l app=goreplay -o name 2>/dev/null) || {
+  echo "ERROR: Failed to list pods with label app=goreplay"
+  exit 1
+}
+echo "Found pods:"
+echo "$pods"
+echo
+
+########################################
+# 4. For each pod, gather logs, describe, and get events
+########################################
+for pod in $pods; do
+  # pod looks like "pod/goreplay-daemon-xyz"
+  pod_name="${pod##*/}"  # remove "pod/" prefix
+
+  echo "=================================================="
+  echo "LOGS for pod: ${pod_name}"
+  echo "=================================================="
+  run_cmd "$KUBECTL logs ${pod_name} -n goreplay" || {
+    echo "WARNING: Failed to get logs for pod ${pod_name}"
+  }
+
+  echo
+  echo "--------------------------------------------------"
+  echo "DESCRIBE for pod: ${pod_name}"
+  echo "--------------------------------------------------"
+  run_cmd "$KUBECTL describe pod -n goreplay ${pod_name}" || {
+    echo "WARNING: Failed to describe pod ${pod_name}"
+  }
+
+  echo
+  echo "--------------------------------------------------"
+  echo "EVENTS for pod: ${pod_name}"
+  echo "--------------------------------------------------"
+  run_cmd "$KUBECTL get events -n goreplay --field-selector involvedObject.name=${pod_name}" || {
+    echo "WARNING: Failed to get events for pod ${pod_name}"
+  }
+  echo
+done
+
+echo "=================================================="
+echo "Telemetry collection complete."
+echo "=================================================="

k8s/goreplay.yaml

@@ -1,15 +1,22 @@
 apiVersion: apps/v1
 kind: DaemonSet
 metadata:
-  name: goreplay-daemon
+  name: goreplay-daemon
 spec:
-  template:
-    spec:
+  selector:
+    matchLabels:
+      app: goreplay
+  template:
+    metadata:
+      labels:
+        app: goreplay
+    spec:
       hostNetwork: true
       serviceAccountName: goreplay
-      containers:
-      - name: goreplay
-        image: buger/goreplay:2.0.0-rc2
-        command:
-          - "--input-raw k8s://deployments/nginx:80"
-          - "--output-stdout"
+      containers:
+      - name: goreplay
+        image: buger/gor:v2.0.0-rc4
+        args:
+        - "--input-raw k8s://deployments/nginx:80"
+        - "--output-stdout"
+        - "--verbose"

kafka.go

@@ -31,6 +31,7 @@ type InputKafkaConfig struct {
 	Host       string `json:"input-kafka-host"`
 	Topic      string `json:"input-kafka-topic"`
 	UseJSON    bool   `json:"input-kafka-json-format"`
+	Offset     string  `json:"input-kafka-offset"`
 	SASLConfig SASLKafkaConfig
 }
 

limiter.go

@@ -31,6 +31,22 @@ func parseLimitOptions(options string) (limit int, isPercent bool) {
 	return
 }
 
+func newLimiterExceptions(l *Limiter) {
+
+	if !l.isPercent {
+		return
+	}
+	speedFactor := float64(l.limit) / float64(100)
+
+	// FileInput、KafkaInput have its own rate limiting. Unlike other inputs we not just dropping requests, we can slow down or speed up request emittion.
+	switch input := l.plugin.(type) {
+	case *FileInput:
+		input.speedFactor = speedFactor
+	case *KafkaInput:
+		input.speedFactor = speedFactor
+	}
+}
+
 // NewLimiter constructor for Limiter, accepts plugin and options
 // `options` allow to sprcify relatve or absolute limiting
 func NewLimiter(plugin interface{}, options string) PluginReadWriter {
@@ -39,17 +55,28 @@ func NewLimiter(plugin interface{}, options string) PluginReadWriter {
 	l.plugin = plugin
 	l.currentTime = time.Now().UnixNano()
 
-	// FileInput have its own rate limiting. Unlike other inputs we not just dropping requests, we can slow down or speed up request emittion.
-	if fi, ok := l.plugin.(*FileInput); ok && l.isPercent {
-		fi.speedFactor = float64(l.limit) / float64(100)
-	}
+	newLimiterExceptions(l)
 
 	return l
 }
 
+func (l *Limiter) isLimitedExceptions() bool {
+	if !l.isPercent {
+		return false
+	}
+	// Fileinput、Kafkainput have its own limiting algorithm
+	switch l.plugin.(type) {
+	case *FileInput:
+		return true
+	case *KafkaInput:
+		return true
+	default:
+		return false
+	}
+}
+
 func (l *Limiter) isLimited() bool {
-	// File input have its own limiting algorithm
-	if _, ok := l.plugin.(*FileInput); ok && l.isPercent {
+	if l.isLimitedExceptions() {
 		return false
 	}
 

middleware/README.md

@@ -117,6 +117,115 @@ Package expose following functions to process raw HTTP payloads:
 Also it is totally legit to use standard `Buffer` functions like `indexOf` for processing the HTTP payload. Just do not forget that if you modify the body, update the `Content-Length` header with a new value. And if you modify any of the headers, line endings should be `\r\n`. Rest is up to your imagination.
 
 
+## Masking PII Data
+
+This middleware provides functionality to mask Personally Identifiable Information (PII) data in HTTP requests based on specified headers and JSON paths. It allows you to define a configuration object that specifies which headers and JSON fields should be masked and the type of data they represent.
+
+```javascript
+const gor = require("goreplay_middleware");
+const faker = require("faker");
+
+// Initialize the middleware
+gor.init();
+
+// Configuration for masking PII data
+const maskConfig = {
+  headers: [
+    { name: "Authorization", type: "token" },
+    { name: "X-API-Key", type: "token" },
+    { name: "X-User-Email", type: "email" },
+    { name: "X-User-Name", type: "name" },
+  ],
+  jsonPaths: [
+    { path: "$.user.email", type: "email" },
+    { path: "$.user.name", type: "name" },
+    { path: "$.user.phone", type: "phone" },
+    { path: "$.user.address", type: "address" },
+  ],
+};
+
+// Function to mask a value based on its type
+function maskValue(type) {
+  switch (type) {
+    case "email":
+      return faker.internet.email();
+    case "name":
+      return faker.name.findName();
+    case "phone":
+      return faker.phone.phoneNumber();
+    case "address":
+      return faker.address.streetAddress();
+    case "token":
+      return faker.random.alphaNumeric(32);
+    default:
+      return "***";
+  }
+}
+
+// Middleware function to mask PII data
+gor.on("message", (data) => {
+  // Mask headers
+  maskConfig.headers.forEach((header) => {
+    const value = gor.httpHeader(data.http, header.name);
+    if (value) {
+      data.http = gor.setHttpHeader(data.http, header.name, maskValue(header.type));
+    }
+  });
+
+  // Mask JSON fields
+  const body = gor.httpBody(data.http);
+  if (body) {
+    try {
+      const jsonBody = JSON.parse(body.toString());
+      maskConfig.jsonPaths.forEach((field) => {
+        const value = eval(`jsonBody${field.path.slice(1)}`);
+        if (value) {
+          eval(`jsonBody${field.path.slice(1)} = maskValue(field.type)`);
+        }
+      });
+      data.http = gor.setHttpBody(data.http, Buffer.from(JSON.stringify(jsonBody)));
+    } catch (error) {
+      console.error("Error parsing JSON body:", error);
+    }
+  }
+
+  return data;
+});
+```
+
+### Configuration
+
+The `maskConfig` object is used to configure the masking behavior. It consists of two properties:
+
+- `headers`: An array of objects representing the headers to be masked. Each object should have the following properties:
+  - `name`: The name of the header.
+  - `type`: The type of data the header represents (e.g., "email", "name", "token").
+
+- `jsonPaths`: An array of objects representing the JSON paths to be masked. Each object should have the following properties:
+  - `path`: The JSON path to the field to be masked (e.g., "$.user.email").
+  - `type`: The type of data the field represents (e.g., "email", "name", "phone", "address").
+
+### Masking Function
+
+The `maskValue` function is responsible for generating masked values based on the data type. It uses the Faker library to generate realistic-looking masked data for different types such as email, name, phone, address, and token. You can extend this function to support additional data types or customize the masking behavior.
+
+### Middleware Function
+
+The middleware function is triggered for each HTTP message (request or response) processed by GoReplay. It performs the following steps:
+
+1. Iterate over the specified headers in the `maskConfig` and mask their values using the `maskValue` function based on the associated data type.
+
+2. Parse the JSON body of the request (if present) and iterate over the specified JSON paths in the `maskConfig`. If a value exists at a given path, replace it with a masked value generated by the `maskValue` function based on the associated data type.
+
+3. Update the request body with the masked JSON data.
+
+4. Return the modified HTTP message.
+
+Note: The middleware uses the `eval` function to dynamically access and modify the JSON object based on the provided paths. Exercise caution when using `eval` and ensure that the paths are properly validated to prevent potential security risks.
+
+To use this middleware, make sure to install the required dependencies (`goreplay_middleware` and `faker`), configure the `maskConfig` object according to your needs, and run the middleware with GoReplay.
+
+
 ## Support
 
 Feel free to ask questions here and by sending email to [support@goreplay.org](mailto:support@goreplay.org). Commercial support is available and welcomed 🙈.

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
+	return &BinaryOutput{address: address}
 }
 
-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
-		}
-	}
-}
-
-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
 }

output_binary_pro.go

@@ -0,0 +1,178 @@
+//go:build pro
+
+package goreplay
+
+import (
+	"sync/atomic"
+	"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"`
+	Timeout        time.Duration `json:"output-binary-timeout"`
+	BufferSize     size.Size     `json:"output-tcp-response-buffer"`
+	Debug          bool          `json:"output-binary-debug"`
+	TrackResponses bool          `json:"output-binary-track-response"`
+}
+
+// BinaryOutput plugin manage pool of workers which send request to replayed server
+// 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
+}
+
+// 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
+		}
+	}
+}
+
+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
+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
+}
+
+// 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()}
+	}
+}
+
+func (o *BinaryOutput) String() string {
+	return "Binary output: " + o.address
+}
+
+// Close closes this plugin for reading
+func (o *BinaryOutput) Close() error {
+	close(o.quit)
+	return nil
+}

output_http.go

@@ -5,7 +5,6 @@ import (
 	"bytes"
 	"crypto/tls"
 	"fmt"
-	"github.com/buger/goreplay/internal/size"
 	"log"
 	"math"
 	"net/http"
@@ -13,6 +12,8 @@ import (
 	"net/url"
 	"sync/atomic"
 	"time"
+
+	"github.com/buger/goreplay/internal/size"
 )
 
 const (
@@ -30,38 +31,44 @@ 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"`
-	WorkerTimeout  time.Duration `json:"output-http-worker-timeout"`
-	BufferSize     size.Size     `json:"output-http-response-buffer"`
-	SkipVerify     bool          `json:"output-http-skip-verify"`
-	rawURL         string
-	url            *url.URL
+	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"`
+	WorkerTimeout     time.Duration `json:"output-http-worker-timeout"`
+	BufferSize        size.Size     `json:"output-http-response-buffer"`
+	SkipVerify        bool          `json:"output-http-skip-verify"`
+	CompatibilityMode bool          `json:"output-http-compatibility-mode"`
+	RequestGroup      string        `json:"output-http-request-group"`
+	Debug             bool          `json:"output-http-debug"`
+	rawURL            string
+	url               *url.URL
 }
 
 func (hoc *HTTPOutputConfig) Copy() *HTTPOutputConfig {
 	return &HTTPOutputConfig{
-		TrackResponses: hoc.TrackResponses,
-		Stats:          hoc.Stats,
-		OriginalHost:   hoc.OriginalHost,
-		RedirectLimit:  hoc.RedirectLimit,
-		WorkersMin:     hoc.WorkersMin,
-		WorkersMax:     hoc.WorkersMax,
-		StatsMs:        hoc.StatsMs,
-		QueueLen:       hoc.QueueLen,
-		ElasticSearch:  hoc.ElasticSearch,
-		Timeout:        hoc.Timeout,
-		WorkerTimeout:  hoc.WorkerTimeout,
-		BufferSize:     hoc.BufferSize,
-		SkipVerify:     hoc.SkipVerify,
+		TrackResponses:    hoc.TrackResponses,
+		Stats:             hoc.Stats,
+		OriginalHost:      hoc.OriginalHost,
+		RedirectLimit:     hoc.RedirectLimit,
+		WorkersMin:        hoc.WorkersMin,
+		WorkersMax:        hoc.WorkersMax,
+		StatsMs:           hoc.StatsMs,
+		QueueLen:          hoc.QueueLen,
+		ElasticSearch:     hoc.ElasticSearch,
+		Timeout:           hoc.Timeout,
+		WorkerTimeout:     hoc.WorkerTimeout,
+		BufferSize:        hoc.BufferSize,
+		SkipVerify:        hoc.SkipVerify,
+		CompatibilityMode: hoc.CompatibilityMode,
+		RequestGroup:      hoc.RequestGroup,
+		Debug:             hoc.Debug,
 	}
 }
 
@@ -69,15 +76,49 @@ func (hoc *HTTPOutputConfig) Copy() *HTTPOutputConfig {
 // By default workers pool is dynamic and starts with 1 worker or workerMin workers
 // You can specify maximum number of workers using `--output-http-workers`
 type HTTPOutput struct {
-	activeWorkers int32
-	config        *HTTPOutputConfig
-	queueStats    *GorStat
-	elasticSearch *ESPlugin
-	client        *HTTPClient
-	stopWorker    chan struct{}
-	queue         chan *Message
-	responses     chan *response
-	stop          chan bool // Channel used only to indicate goroutine should shutdown
+	activeWorkers  int64
+	config         *HTTPOutputConfig
+	queueStats     *GorStat
+	elasticSearch  *ESPlugin
+	client         *HTTPClient
+	stopWorker     chan struct{}
+	queue          chan *Message
+	responses      chan *response
+	stop           chan bool // Channel used only to indicate goroutine should shutdown
+	workerSessions map[string]*httpWorker
+}
+
+type httpWorker struct {
+	output       *HTTPOutput
+	client       *HTTPClient
+	lastActivity time.Time
+	queue        chan *Message
+	stop         chan bool
+}
+
+func newHTTPWorker(output *HTTPOutput, queue chan *Message) *httpWorker {
+	client := NewHTTPClient(output.config)
+
+	w := &httpWorker{client: client, output: output}
+	if queue == nil {
+		w.queue = make(chan *Message, 100)
+	} else {
+		w.queue = queue
+	}
+	w.stop = make(chan bool)
+
+	go func() {
+		for {
+			select {
+			case msg := <-w.queue:
+				output.sendRequest(client, msg)
+			case <-w.stop:
+				return
+			}
+		}
+	}()
+
+	return w
 }
 
 // NewHTTPOutput constructor for HTTPOutput
@@ -107,7 +148,7 @@ func NewHTTPOutput(address string, config *HTTPOutputConfig) PluginReadWriter {
 		newConfig.WorkersMin = 1000
 	}
 	if newConfig.WorkersMax <= 0 {
-		newConfig.WorkersMax = math.MaxInt32 // idealy so large
+		newConfig.WorkersMax = math.MaxInt32 // ideally so large
 	}
 	if newConfig.WorkersMax < newConfig.WorkersMin {
 		newConfig.WorkersMax = newConfig.WorkersMin
@@ -139,11 +180,18 @@ func NewHTTPOutput(address string, config *HTTPOutputConfig) PluginReadWriter {
 		o.elasticSearch.Init(o.config.ElasticSearch)
 	}
 	o.client = NewHTTPClient(o.config)
-	o.activeWorkers += int32(o.config.WorkersMin)
-	for i := 0; i < o.config.WorkersMin; i++ {
-		go o.startWorker()
+
+	if Settings.RecognizeTCPSessions {
+		o.workerSessions = make(map[string]*httpWorker, 100)
+		go o.sessionWorkerMaster()
+	} else {
+		o.activeWorkers += int64(o.config.WorkersMin)
+		for i := 0; i < o.config.WorkersMin; i++ {
+			go o.startWorker()
+		}
+		go o.workerMaster()
 	}
-	go o.workerMaster()
+
 	return o
 }
 
@@ -164,16 +212,48 @@ func (o *HTTPOutput) workerMaster() {
 		}
 		// rollback workers
 	rollback:
-		if atomic.LoadInt32(&o.activeWorkers) > int32(o.config.WorkersMin) && len(o.queue) < 1 {
+		if atomic.LoadInt64(&o.activeWorkers) > int64(o.config.WorkersMin) && len(o.queue) < 1 {
 			// close one worker
 			o.stopWorker <- struct{}{}
-			atomic.AddInt32(&o.activeWorkers, -1)
+			atomic.AddInt64(&o.activeWorkers, -1)
 			goto rollback
 		}
 		timer.Reset(o.config.WorkerTimeout)
 	}
 }
 
+func (o *HTTPOutput) sessionWorkerMaster() {
+	gc := time.Tick(time.Second)
+
+	for {
+		select {
+		case msg := <-o.queue:
+			id := payloadID(msg.Meta)
+			sessionID := string(id[0:20])
+			worker, ok := o.workerSessions[sessionID]
+
+			if !ok {
+				atomic.AddInt64(&o.activeWorkers, 1)
+				worker = newHTTPWorker(o, nil)
+				o.workerSessions[sessionID] = worker
+			}
+
+			worker.queue <- msg
+			worker.lastActivity = time.Now()
+		case <-gc:
+			now := time.Now()
+
+			for id, w := range o.workerSessions {
+				if !w.lastActivity.IsZero() && now.Sub(w.lastActivity) >= 120*time.Second {
+					w.stop <- true
+					delete(o.workerSessions, id)
+					atomic.AddInt64(&o.activeWorkers, -1)
+				}
+			}
+		}
+	}
+}
+
 func (o *HTTPOutput) startWorker() {
 	for {
 		select {
@@ -200,13 +280,25 @@ func (o *HTTPOutput) PluginWrite(msg *Message) (n int, err error) {
 	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) {
-			go o.startWorker()
-			atomic.AddInt32(&o.activeWorkers, 1)
+
+	if !Settings.RecognizeTCPSessions && o.config.WorkersMax != o.config.WorkersMin {
+		workersCount := int(atomic.LoadInt64(&o.activeWorkers))
+
+		if len(o.queue) > workersCount {
+			extraWorkersReq := len(o.queue) - workersCount + 1
+			maxWorkersAvailable := o.config.WorkersMax - workersCount
+			if extraWorkersReq > maxWorkersAvailable {
+				extraWorkersReq = maxWorkersAvailable
+			}
+			if extraWorkersReq > 0 {
+				for i := 0; i < extraWorkersReq; i++ {
+					go o.startWorker()
+					atomic.AddInt64(&o.activeWorkers, 1)
+				}
+			}
 		}
 	}
+
 	return len(msg.Data) + len(msg.Meta), nil
 }
 
@@ -292,7 +384,7 @@ func NewHTTPClient(config *HTTPOutputConfig) *HTTPClient {
 		},
 	}
 	if config.SkipVerify {
-		// clone to avoid modying global default RoundTripper
+		// clone to avoid modifying global default RoundTripper
 		transport = http.DefaultTransport.(*http.Transport).Clone()
 		transport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true}
 		client.Client.Transport = transport
@@ -301,7 +393,7 @@ func NewHTTPClient(config *HTTPOutputConfig) *HTTPClient {
 	return client
 }
 
-// Send sends an http request using client create by NewHTTPClient
+// Send sends an http request using client created by NewHTTPClient
 func (c *HTTPClient) Send(data []byte) ([]byte, error) {
 	var req *http.Request
 	var resp *http.Response

output_s3.go

@@ -1,131 +1,29 @@
+//go:build !pro
+
 package goreplay
 
 import (
-	_ "bufio"
+	"errors"
 	"fmt"
-	_ "io"
-	"log"
-	"math/rand"
-	"os"
-	"path/filepath"
-	"strings"
-
-	"github.com/aws/aws-sdk-go/aws"
-	"github.com/aws/aws-sdk-go/aws/session"
-	"github.com/aws/aws-sdk-go/service/s3"
-	_ "github.com/aws/aws-sdk-go/service/s3/s3manager"
 )
 
 // S3Output output plugin
-type S3Output struct {
-	pathTemplate string
-
-	buffer  *FileOutput
-	session *session.Session
-	config  *FileOutputConfig
-	closeCh chan struct{}
-}
+type S3Output struct{}
 
 // NewS3Output constructor for FileOutput, accepts path
 func NewS3Output(pathTemplate string, config *FileOutputConfig) *S3Output {
-	if !PRO {
-		log.Fatal("Using S3 output and input requires PRO license")
-		return nil
-	}
-
-	o := new(S3Output)
-	o.pathTemplate = pathTemplate
-	o.config = config
-	o.config.onClose = o.onBufferUpdate
-
-	if config.BufferPath == "" {
-		config.BufferPath = "/tmp"
-	}
-
-	rnd := rand.Int63()
-	bufferName := fmt.Sprintf("gor_output_s3_%d_buf_", rnd)
-
-	pathParts := strings.Split(pathTemplate, "/")
-	bufferName += pathParts[len(pathParts)-1]
-
-	if strings.HasSuffix(o.pathTemplate, ".gz") {
-		bufferName += ".gz"
-	}
-
-	bufferPath := filepath.Join(config.BufferPath, bufferName)
-
-	o.buffer = NewFileOutput(bufferPath, config)
-	o.connect()
-
-	return o
+	fmt.Println("S3 output is only available in the pro version")
+	return &S3Output{}
 }
 
-func (o *S3Output) connect() {
-	if o.session == nil {
-		o.session = session.Must(session.NewSession(awsConfig()))
-		log.Println("[S3 Output] S3 connection successfully initialized")
-	}
-}
-
-// PluginWrite writes message to this plugin
 func (o *S3Output) PluginWrite(msg *Message) (n int, err error) {
-	return o.buffer.PluginWrite(msg)
+	return 0, errors.New("S3 output is only available in the pro version")
 }
 
 func (o *S3Output) String() string {
-	return "S3 output: " + o.pathTemplate
+	return "S3 output (pro version only)"
 }
 
-// Close close the buffer of the S3 connection
 func (o *S3Output) Close() error {
-	return o.buffer.Close()
-}
-
-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 (o *S3Output) keyPath(idx int) (bucket, key string) {
-	bucket, key = parseS3Url(o.pathTemplate)
-
-	for name, fn := range dateFileNameFuncs {
-		key = strings.Replace(key, name, fn(o.buffer), -1)
-	}
-
-	key = setFileIndex(key, idx)
-
-	return
-}
-
-func (o *S3Output) onBufferUpdate(path string) {
-	svc := s3.New(o.session)
-	idx := getFileIndex(path)
-	bucket, key := o.keyPath(idx)
-
-	file, err := os.Open(path)
-	if err != nil {
-		Debug(0, fmt.Sprintf("[S3 Output] Failed to open file %q. err: %q", path, err))
-		return
-	}
-	defer os.Remove(path)
-
-	_, err = svc.PutObject(&s3.PutObjectInput{
-		Body:   file,
-		Bucket: aws.String(bucket),
-		Key:    aws.String(key),
-	})
-	if err != nil {
-		Debug(0, fmt.Sprintf("[S3 Output] Failed to upload data to %q/%q, %q", bucket, key, err))
-		return
-	}
-
-	if o.closeCh != nil {
-		o.closeCh <- struct{}{}
-	}
+	return errors.New("S3 output is only available in the pro version")
 }

output_s3_pro.go

@@ -0,0 +1,117 @@
+//go:build pro
+
+package goreplay
+
+import (
+	_ "bufio"
+	"fmt"
+	_ "io"
+	"log"
+	"math/rand"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/s3"
+	_ "github.com/aws/aws-sdk-go/service/s3/s3manager"
+)
+
+var _ PluginWriter = (*S3Output)(nil)
+
+// S3Output output plugin
+type S3Output struct {
+	pathTemplate string
+
+	buffer  *FileOutput
+	session *session.Session
+	config  *FileOutputConfig
+	closeC  chan struct{}
+}
+
+// NewS3Output constructor for FileOutput, accepts path
+func NewS3Output(pathTemplate string, config *FileOutputConfig) *S3Output {
+	o := new(S3Output)
+	o.pathTemplate = pathTemplate
+	o.config = config
+	o.config.onClose = o.onBufferUpdate
+
+	if config.BufferPath == "" {
+		config.BufferPath = "/tmp"
+	}
+
+	rnd := rand.Int63()
+	buffer_name := fmt.Sprintf("gor_output_s3_%d_buf_", rnd)
+
+	pathParts := strings.Split(pathTemplate, "/")
+	buffer_name += pathParts[len(pathParts)-1]
+
+	if strings.HasSuffix(o.pathTemplate, ".gz") {
+		buffer_name += ".gz"
+	}
+
+	buffer_path := filepath.Join(config.BufferPath, buffer_name)
+
+	o.buffer = NewFileOutput(buffer_path, config)
+	o.connect()
+
+	return o
+}
+
+func (o *S3Output) connect() {
+	if o.session == nil {
+		o.session = session.Must(session.NewSession(awsConfig()))
+		log.Println("[S3 Output] S3 connection succesfully initialized")
+	}
+}
+
+func (o *S3Output) PluginWrite(msg *Message) (n int, err error) {
+	return o.buffer.PluginWrite(msg)
+}
+
+func (o *S3Output) String() string {
+	return "S3 output: " + o.pathTemplate
+}
+
+func (o *S3Output) Close() error {
+	return o.buffer.Close()
+}
+
+func (o *S3Output) keyPath(idx int) (bucket, key string) {
+	bucket, key = parseS3Url(o.pathTemplate)
+
+	for name, fn := range dateFileNameFuncs {
+		key = strings.Replace(key, name, fn(o.buffer), -1)
+	}
+
+	key = setFileIndex(key, idx)
+
+	return
+}
+
+func (o *S3Output) onBufferUpdate(path string) {
+	svc := s3.New(o.session)
+	idx := getFileIndex(path)
+	bucket, key := o.keyPath(idx)
+
+	file, _ := os.Open(path)
+	// reader := bufio.NewReader(file)
+
+	_, err := svc.PutObject(&s3.PutObjectInput{
+		Body:   file,
+		Bucket: aws.String(bucket),
+		Key:    aws.String(key),
+	})
+	if err != nil {
+		log.Printf("[S3 Output] Failed to upload data to %s/%s, %s\n", bucket, key, err)
+		os.Remove(path)
+		return
+	}
+
+	os.Remove(path)
+
+	if o.closeC != nil {
+		o.closeC <- struct{}{}
+	}
+}

plugins.go

@@ -159,7 +159,7 @@ func NewPlugins() *InOutPlugins {
 	}
 
 	if Settings.InputKafkaConfig.Host != "" && Settings.InputKafkaConfig.Topic != "" {
-		plugins.registerPlugin(NewKafkaInput, "", &Settings.InputKafkaConfig, &Settings.KafkaTLSConfig)
+		plugins.registerPlugin(NewKafkaInput, Settings.InputKafkaConfig.Offset, &Settings.InputKafkaConfig, &Settings.KafkaTLSConfig)
 	}
 
 	return plugins

pro.go

@@ -1,5 +1,10 @@
+//go:build pro
+
 package goreplay
 
 // PRO this value indicates if goreplay is running in PRO mode..
 // it must not be modified explicitly in production
-var PRO = false
+var PRO = true
+
+// SettingsHook is intentionally left as a no-op
+var SettingsHook = func(*AppSettings) {}

settings.go

@@ -3,11 +3,12 @@ package goreplay
 import (
 	"flag"
 	"fmt"
-	"github.com/buger/goreplay/internal/size"
 	"os"
 	"strconv"
 	"sync"
 	"time"
+
+	"github.com/buger/goreplay/internal/size"
 )
 
 // DEMO indicates that goreplay is running in demo mode
@@ -250,6 +251,7 @@ func init() {
 	flag.StringVar(&Settings.InputKafkaConfig.SASLConfig.Mechanism, "input-kafka-mechanism", "", "mechanism\n\tgor --input-raw :8080 --output-kafka-mechanism 'SCRAM-SHA-512'")
 	flag.StringVar(&Settings.InputKafkaConfig.SASLConfig.Username, "input-kafka-username", "", "username\n\tgor --input-raw :8080 --output-kafka-username 'username'")
 	flag.StringVar(&Settings.InputKafkaConfig.SASLConfig.Password, "input-kafka-password", "", "password\n\tgor --input-raw :8080 --output-kafka-password 'password'")
+	flag.StringVar(&Settings.InputKafkaConfig.Offset, "input-kafka-offset", "-1", "Specify offset in Kafka partitions start to consume\n\t-1: Starts from newest, -2: Starts from oldest\nAnd supported for showdown or speedup for emitting!\n\tgor --input-kafka-offset \"-2|200%\"")
 
 	flag.StringVar(&Settings.KafkaTLSConfig.CACert, "kafka-tls-ca-cert", "", "CA certificate for Kafka TLS Config:\n\tgor  --input-raw :3000 --output-kafka-host '192.168.0.1:9092' --output-kafka-topic 'topic' --kafka-tls-ca-cert cacert.cer.pem --kafka-tls-client-cert client.cer.pem --kafka-tls-client-key client.key.pem")
 	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)")
@@ -276,6 +278,8 @@ func init() {
 }
 
 func CheckSettings() {
+	SettingsHook(&Settings)
+
 	if Settings.OutputFileConfig.SizeLimit < 1 {
 		Settings.OutputFileConfig.SizeLimit.Set("32mb")
 	}

version.go

@@ -1,4 +1,4 @@
 package goreplay
 
 // VERSION the current version of goreplay
-var VERSION = "1.3.0"
+var VERSION = "2.0.0"