Added support for setting seed for deterministic results.

README.md

@@ -56,17 +56,18 @@ stopping you from creating a client in any other language (see
       - [latency](#latency)
       - [down](#down)
       - [bandwidth](#bandwidth)
-      - [slow_close](#slow_close)
+      - [slow\_close](#slow_close)
       - [timeout](#timeout)
-      - [reset_peer](#reset_peer)
+      - [reset\_peer](#reset_peer)
       - [slicer](#slicer)
-      - [limit_data](#limit_data)
+      - [limit\_data](#limit_data)
     - [HTTP API](#http-api)
       - [Proxy fields:](#proxy-fields)
       - [Toxic fields:](#toxic-fields)
       - [Endpoints](#endpoints)
       - [Populating Proxies](#populating-proxies)
     - [CLI Example](#cli-example)
+    - [Deterministic results](#deterministic-results)
     - [Metrics](#metrics)
     - [Frequently Asked Questions](#frequently-asked-questions)
     - [Development](#development)
@@ -571,6 +572,10 @@ $ redis-cli -p 26379
 Could not connect to Redis at 127.0.0.1:26379: Connection refused
 ```
 
+### Deterministic results
+
+To achieve deterministic results (for example to replicate the issue observed in some e2e tests) pass `--seed <seed>` to toxiproxy-cli. This will ensure that the random number generator is seeded with the same value, resulting in the same sequence of random numbers being generated.
+
 ### Metrics
 
 Toxiproxy exposes Prometheus-compatible metrics via its HTTP API at /metrics.

api.go

@@ -35,18 +35,20 @@ type ApiServer struct {
 	Metrics    *metricsContainer
 	Logger     *zerolog.Logger
 	http       *http.Server
+	seed       int64
 }
 
 const (
 	wait_timeout = 30 * time.Second
 	read_timeout = 15 * time.Second
 )
 
-func NewServer(m *metricsContainer, logger zerolog.Logger) *ApiServer {
+func NewServer(m *metricsContainer, logger zerolog.Logger, seed int64) *ApiServer {
 	return &ApiServer{
 		Collection: NewProxyCollection(),
 		Metrics:    m,
 		Logger:     &logger,
+		seed:       seed,
 	}
 }
 

api_test.go

@@ -32,6 +32,7 @@ func WithServer(t *testing.T, f func(string)) {
 		testServer = toxiproxy.NewServer(
 			toxiproxy.NewMetricsContainer(prometheus.NewRegistry()),
 			log,
+			time.Now().UnixNano(),
 		)
 
 		go testServer.Listen("localhost:8475")

cmd/server/server.go

@@ -67,6 +67,7 @@ func run() error {
 		return nil
 	}
 
+	seed := cli.seed
 	rand.New(rand.NewSource(cli.seed)) // #nosec G404 -- ignoring this rule
 
 	logger := setupLogger()
@@ -78,7 +79,7 @@ func run() error {
 		Msg("Starting Toxiproxy")
 
 	metrics := toxiproxy.NewMetricsContainer(prometheus.NewRegistry())
-	server := toxiproxy.NewServer(metrics, logger)
+	server := toxiproxy.NewServer(metrics, logger, seed)
 	if cli.proxyMetrics {
 		server.Metrics.ProxyMetrics = collectors.NewProxyMetricCollectors()
 	}

link.go

@@ -106,7 +106,7 @@ func (link *ToxicLink) Start(
 			}
 		}
 
-		go link.stubs[i].Run(toxic)
+		go link.stubs[i].Run(toxic, server.seed)
 	}
 
 	go link.write(labels, name, server, dest)
@@ -182,8 +182,8 @@ func (link *ToxicLink) AddToxic(toxic *toxics.ToxicWrapper) {
 			link.stubs[i].State = stateful.NewState()
 		}
 
-		go link.stubs[i].Run(toxic)
-		go link.stubs[i-1].Run(link.toxics.chain[link.direction][i-1])
+		go link.stubs[i].Run(toxic, link.proxy.apiServer.seed)
+		go link.stubs[i-1].Run(link.toxics.chain[link.direction][i-1], link.proxy.apiServer.seed)
 	} else {
 		// This link is already closed, make sure the new toxic matches
 		link.stubs[i].Output = newin // The real output is already closed, close this instead
@@ -194,7 +194,7 @@ func (link *ToxicLink) AddToxic(toxic *toxics.ToxicWrapper) {
 // Update an existing toxic in the chain.
 func (link *ToxicLink) UpdateToxic(toxic *toxics.ToxicWrapper) {
 	if link.stubs[toxic.Index].InterruptToxic() {
-		go link.stubs[toxic.Index].Run(toxic)
+		go link.stubs[toxic.Index].Run(toxic, link.proxy.apiServer.seed)
 	}
 }
 
@@ -273,7 +273,8 @@ func (link *ToxicLink) RemoveToxic(ctx context.Context, toxic *toxics.ToxicWrapp
 		link.stubs[toxic_index-1].Output = link.stubs[toxic_index].Output
 		link.stubs = append(link.stubs[:toxic_index], link.stubs[toxic_index+1:]...)
 
-		go link.stubs[toxic_index-1].Run(link.toxics.chain[link.direction][toxic_index-1])
+		go link.stubs[toxic_index-1].Run(link.toxics.chain[link.direction][toxic_index-1],
+			link.proxy.apiServer.seed)
 	}
 }
 

link_test.go

@@ -86,8 +86,12 @@ func TestStubInitializaationWithToxics(t *testing.T) {
 func TestAddRemoveStubs(t *testing.T) {
 	ctx := context.Background()
 	collection := NewToxicCollection(nil)
-	link := NewToxicLink(nil, collection, stream.Downstream, zerolog.Nop())
-	go link.stubs[0].Run(collection.chain[stream.Downstream][0])
+	dummyProxy := NewProxy(NewServer(nil, zerolog.Nop(), time.Now().UnixNano()),
+		"DummyProxy",
+		"localhost:0",
+		"upstream")
+	link := NewToxicLink(dummyProxy, collection, stream.Downstream, zerolog.Nop())
+	go link.stubs[0].Run(collection.chain[stream.Downstream][0], link.proxy.apiServer.seed)
 	collection.links["test"] = link
 
 	// Add stubs
@@ -140,8 +144,12 @@ func TestAddRemoveStubs(t *testing.T) {
 func TestNoDataDropped(t *testing.T) {
 	ctx := context.Background()
 	collection := NewToxicCollection(nil)
-	link := NewToxicLink(nil, collection, stream.Downstream, zerolog.Nop())
-	go link.stubs[0].Run(collection.chain[stream.Downstream][0])
+	dummyProxy := NewProxy(NewServer(nil, zerolog.Nop(), time.Now().UnixNano()),
+		"DummyProxy",
+		"localhost:0",
+		"upstream")
+	link := NewToxicLink(dummyProxy, collection, stream.Downstream, zerolog.Nop())
+	go link.stubs[0].Run(collection.chain[stream.Downstream][0], link.proxy.apiServer.seed)
 	collection.links["test"] = link
 
 	toxic := &toxics.ToxicWrapper{
@@ -196,8 +204,12 @@ func TestNoDataDropped(t *testing.T) {
 
 func TestToxicity(t *testing.T) {
 	collection := NewToxicCollection(nil)
-	link := NewToxicLink(nil, collection, stream.Downstream, zerolog.Nop())
-	go link.stubs[0].Run(collection.chain[stream.Downstream][0])
+	dummyProxy := NewProxy(NewServer(nil, zerolog.Nop(), time.Now().UnixNano()),
+		"DummyProxy",
+		"localhost:0",
+		"upstream")
+	link := NewToxicLink(dummyProxy, collection, stream.Downstream, zerolog.Nop())
+	go link.stubs[0].Run(collection.chain[stream.Downstream][0], link.proxy.apiServer.seed)
 	collection.links["test"] = link
 
 	toxic := &toxics.ToxicWrapper{
@@ -247,9 +259,10 @@ func TestStateCreated(t *testing.T) {
 	if flag.Lookup("test.v").DefValue == "true" {
 		log = zerolog.New(os.Stdout).With().Caller().Timestamp().Logger()
 	}
-
-	link := NewToxicLink(nil, collection, stream.Downstream, log)
-	go link.stubs[0].Run(collection.chain[stream.Downstream][0])
+	dummyServer := NewServer(nil, log, time.Now().UnixNano())
+	dummyProxy := NewProxy(dummyServer, "DummyProxy", "localhost:0", "upstream")
+	link := NewToxicLink(dummyProxy, collection, stream.Downstream, log)
+	go link.stubs[0].Run(collection.chain[stream.Downstream][0], link.proxy.apiServer.seed)
 	collection.links["test"] = link
 
 	collection.chainAddToxic(&toxics.ToxicWrapper{
@@ -271,10 +284,11 @@ func TestRemoveToxicWithBrokenConnection(t *testing.T) {
 		log = zerolog.New(os.Stdout).With().Caller().Timestamp().Logger()
 	}
 	ctx = log.WithContext(ctx)
-
 	collection := NewToxicCollection(nil)
-	link := NewToxicLink(nil, collection, stream.Downstream, log)
-	go link.stubs[0].Run(collection.chain[stream.Downstream][0])
+	dummyServer := NewServer(nil, log, time.Now().UnixNano())
+	dummyProxy := NewProxy(dummyServer, "DummyProxy", "localhost:0", "upstream")
+	link := NewToxicLink(dummyProxy, collection, stream.Downstream, log)
+	go link.stubs[0].Run(collection.chain[stream.Downstream][0], link.proxy.apiServer.seed)
 	collection.links["test"] = link
 
 	toxics := [2]*toxics.ToxicWrapper{
@@ -323,3 +337,55 @@ func TestRemoveToxicWithBrokenConnection(t *testing.T) {
 	collection.chainRemoveToxic(ctx, toxics[0])
 	collection.chainRemoveToxic(ctx, toxics[1])
 }
+
+func TestStableToxicityWithSeed(t *testing.T) {
+	collection := NewToxicCollection(nil)
+	//for a seed == 1 the random number generated is 0.604(...)
+	dummyProxy := NewProxy(NewServer(nil, zerolog.Nop(), 1),
+		"DummyProxy",
+		"localhost:0",
+		"upstream")
+	link := NewToxicLink(dummyProxy, collection, stream.Downstream, zerolog.Nop())
+	go link.stubs[0].Run(collection.chain[stream.Downstream][0], link.proxy.apiServer.seed)
+	collection.links["test"] = link
+
+	toxic := &toxics.ToxicWrapper{
+		Toxic:     new(toxics.TimeoutToxic),
+		Name:      "timeout1",
+		Type:      "timeout",
+		Direction: stream.Downstream,
+		Toxicity:  0.603,
+	}
+	collection.chainAddToxic(toxic)
+
+	// Toxic should be a Noop because of toxicity
+	n, err := link.input.Write([]byte{42})
+	if n != 1 || err != nil {
+		t.Fatalf("Write failed: %d %v", n, err)
+	}
+	buf := make([]byte, 2)
+	n, err = link.output.Read(buf)
+	if n != 1 || err != nil {
+		t.Fatalf("Read failed: %d %v", n, err)
+	} else if buf[0] != 42 {
+		t.Fatalf("Read wrong byte: %x", buf[0])
+	}
+
+	toxic.Toxicity = 0.605
+	toxic.Toxic.(*toxics.TimeoutToxic).Timeout = 100
+	collection.chainUpdateToxic(toxic)
+
+	err = testhelper.TimeoutAfter(150*time.Millisecond, func() {
+		n, err = link.input.Write([]byte{42})
+		if n != 1 || err != nil {
+			t.Fatalf("Write failed: %d %v", n, err)
+		}
+		n, err = link.output.Read(buf)
+		if n != 0 || err != io.EOF {
+			t.Fatalf("Read did not get EOF: %d %v", n, err)
+		}
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+}

metrics_test.go

@@ -9,6 +9,7 @@ import (
 	"regexp"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/rs/zerolog"
@@ -18,7 +19,7 @@ import (
 )
 
 func TestProxyMetricsReceivedSentBytes(t *testing.T) {
-	srv := NewServer(NewMetricsContainer(prometheus.NewRegistry()), zerolog.Nop())
+	srv := NewServer(NewMetricsContainer(prometheus.NewRegistry()), zerolog.Nop(), time.Now().UnixNano()) //nolint:lll
 	srv.Metrics.ProxyMetrics = collectors.NewProxyMetricCollectors()
 
 	proxy := NewProxy(srv, "test_proxy_metrics_received_sent_bytes", "localhost:0", "upstream")
@@ -55,7 +56,7 @@ func TestProxyMetricsReceivedSentBytes(t *testing.T) {
 }
 
 func TestRuntimeMetricsBuildInfo(t *testing.T) {
-	srv := NewServer(NewMetricsContainer(prometheus.NewRegistry()), zerolog.Nop())
+	srv := NewServer(NewMetricsContainer(prometheus.NewRegistry()), zerolog.Nop(), time.Now().UnixNano()) //nolint:lll
 	srv.Metrics.RuntimeMetrics = collectors.NewRuntimeMetricCollectors()
 
 	expected := `go_build_info{checksum="[^"]*",path="[^"]*",version="[^"]*"} 1`

toxics/toxic.go

@@ -78,10 +78,11 @@ func NewToxicStub(input <-chan *stream.StreamChunk, output chan<- *stream.Stream
 
 // Begin running a toxic on this stub, can be interrupted.
 // Runs a noop toxic randomly depending on toxicity.
-func (s *ToxicStub) Run(toxic *ToxicWrapper) {
+func (s *ToxicStub) Run(toxic *ToxicWrapper, seed int64) {
 	s.running = make(chan struct{})
 	defer close(s.running)
-	randomToxicity := rand.Float32() // #nosec G404 -- was ignored before too
+	r := rand.New(rand.NewSource(seed)) // #nosec G404 -- was ignored before too
+	randomToxicity := r.Float32()       // #nosec G404 -- was ignored before too
 	if randomToxicity < toxic.Toxicity {
 		toxic.Pipe(s)
 	} else {

toxics/toxic_test.go

@@ -32,6 +32,7 @@ func NewTestProxy(name, upstream string) *toxiproxy.Proxy {
 	srv := toxiproxy.NewServer(
 		toxiproxy.NewMetricsContainer(prometheus.NewRegistry()),
 		log,
+		time.Now().UnixNano(),
 	)
 	srv.Metrics.ProxyMetrics = collectors.NewProxyMetricCollectors()
 	proxy := toxiproxy.NewProxy(srv, name, "localhost:0", upstream)

toxiproxy_test.go

@@ -5,6 +5,7 @@ import (
 	"net"
 	"os"
 	"testing"
+	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/rs/zerolog"
@@ -22,6 +23,7 @@ func NewTestProxy(name, upstream string) *toxiproxy.Proxy {
 	srv := toxiproxy.NewServer(
 		toxiproxy.NewMetricsContainer(prometheus.NewRegistry()),
 		log,
+		time.Now().UnixNano(),
 	)
 	srv.Metrics.ProxyMetrics = collectors.NewProxyMetricCollectors()
 	proxy := toxiproxy.NewProxy(srv, name, "localhost:0", upstream)