record-tester: Test stream health (#118)

* cmd: Create healthtester cmd based on recordtester

Basically copied the whole command and then went through it
rewriting some code as a way to understand the logic.

Also copied some helpers from analyzer code (contextUntilSig)

* healthtester: Create helper for running multiple tests

* recordtester: Improve PagerDuty formatting/links

* recordtester: Allow cancelling in between tests

* testers: Create Stream2Options

* testers: Create Finite iface and receive on streamer2

* cmd: Remove healthchecker new command

Can check any code on history if necessary

* testers: Dummy stream health tester for arch review

* DEV-REVERT: Use local livepeer-data lib

* testers: Simplify additional tests glue

* testers: Implement health tester for real

* cmd/recordtester: Plug everything together

* Revert "DEV-REVERT: Use local livepeer-data lib"

This reverts commit b6cb78c8b0ae7e2d2aad575c9695c1d56cc5d640.

* go.mod: Use beta (unmerged) version of livepeer-data

* cmd/recordtester: Simplify analyzer URL flag passing

* testers: Improve logs on stream health tester

* testers: Fix unhealthy timer

* testers: Log all errors after half the wait time

* testers: Add insane log

* record-tester: moar insane logs

* recordtester: Ignore commas in env vars

* Revert "record-tester: moar insane logs"

This reverts commit f516d3f0242cf662b9651c8a5fa00c5032e8f16c.

* testers: Improve formatting of stream health errors

Aggregate regions for each error message and try to
make it clearer that we are testing stream health global
API, not multistream.

* testers: A couple more formatting improvements

* record-tester: Improve pagerduty API usage

Use a dedup key so we dont spam the oncall engineer
and also make sure to resolve incidents automatically
once the tests start working.

* testers: Improve formatting of regions list

Just say all regions when it's all regions

* go.mod: Update analyzer client version

* recordtester: Add option to only send lopri PagerDuty alerts

* recordtester: Include pagerduty component in alert

Will differentiate messages on #incident-response based
on the component, which we'll configure to be GCP/Datapacket.

* health-tester: Make sure to compare health with same time

Otherwise absolutely every error message is going to be slightly
different.

* record-tester: Remove all fatal logs from m3utester2

* record-tester: Use different dedup key when lopri

Otherwise it could get confusing that the incident with
the [LOPRI] prefix sends a high urgency alert.

* record-tester: Make sure to finish on fatal errors

* record-tester: Ensure a global test timeout

* record-tester: Looser global timeout and proper log

* record-tester: Accumulate errors from finite testers

* record-tester: Bring back early test of crt.ctx.Err

Cleaner discord msgs

* stream-tester: Add hostname to start&end logs

cmd/loadtester/loadtester.go

@@ -202,7 +202,7 @@ func main() {
 			}
 			glog.V(model.SHORT).Infof("RTMP: %s", rtmpURL)
 			glog.V(model.SHORT).Infof("MEDIA: %s", mediaURL)
-			sr2 := testers.NewStreamer2(ctx, false, *mist, false, false, false)
+			sr2 := testers.NewStreamer2(ctx, testers.Streamer2Options{MistMode: *mist})
 			go sr2.StartStreaming(sourceFileName, rtmpURL, mediaURL, waitForTarget, timeToStream)
 			go func() {
 				<-sr2.Done()
@@ -225,7 +225,7 @@ func main() {
 				// glog.Error(err)
 				return nil, err
 			}
-			sr2 := testers.NewStreamer2(ctx, false, false, false, false, false)
+			sr2 := testers.NewStreamer2(ctx, testers.Streamer2Options{})
 			go sr2.StartStreaming(sourceFileName, rtmpURL, mediaURL, waitForTarget, timeToStream)
 			return sr2, nil
 		}

cmd/recordtester/recordtester.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/golang/glog"
 	"github.com/livepeer/joy4/format"
+	"github.com/livepeer/livepeer-data/pkg/client"
 	"github.com/livepeer/stream-tester/apis/livepeer"
 	"github.com/livepeer/stream-tester/internal/app/recordtester"
 	"github.com/livepeer/stream-tester/internal/metrics"
@@ -48,15 +49,18 @@ func main() {
 	pauseDuration := fs.Duration("pause-dur", 0, "How long to wait between two consecutive RTMP streams that will comprise one user session")
 	apiToken := fs.String("api-token", "", "Token of the Livepeer API to be used")
 	apiServer := fs.String("api-server", "livepeer.com", "Server of the Livepeer API to be used")
+	analyzerServers := fs.String("analyzer-servers", "", "Comma-separated list of base URLs to connect for the Stream Health Analyzer API (defaults to --api-server)")
 	fileArg := fs.String("file", "bbb_sunflower_1080p_30fps_normal_t02.mp4", "File to stream")
 	continuousTest := fs.Duration("continuous-test", 0, "Do continuous testing")
 	useHttp := fs.Bool("http", false, "Do HTTP tests instead of RTMP")
 	testMP4 := fs.Bool("mp4", false, "Download MP4 of recording")
+	testStreamHealth := fs.Bool("stream-health", false, "Check stream health during test")
 	discordURL := fs.String("discord-url", "", "URL of Discord's webhook to send messages to Discord channel")
 	discordUserName := fs.String("discord-user-name", "", "User name to use when sending messages to Discord")
 	discordUsersToNotify := fs.String("discord-users", "", "Id's of users to notify in case of failure")
 	pagerDutyIntegrationKey := fs.String("pagerduty-integration-key", "", "PagerDuty integration key")
 	pagerDutyComponent := fs.String("pagerduty-component", "", "PagerDuty component")
+	pagerDutyLowUrgency := fs.Bool("pagerduty-low-urgency", false, "Whether to send only low-urgency PagerDuty alerts")
 	bind := fs.String("bind", "0.0.0.0:9090", "Address to bind metric server to")
 
 	_ = fs.String("config", "", "config file (optional)")
@@ -65,6 +69,7 @@ func main() {
 		ff.WithConfigFileFlag("config"),
 		ff.WithConfigFileParser(ff.PlainParser),
 		ff.WithEnvVarPrefix("RT"),
+		ff.WithEnvVarIgnoreCommas(true),
 	)
 	flag.CommandLine.Parse(nil)
 	vFlag.Value.Set(*verbosity)
@@ -78,7 +83,7 @@ func main() {
 	if *version {
 		return
 	}
-	metrics.InitCensus(hostName, model.Version, "loadtester")
+	metrics.InitCensus(hostName, model.Version, "recordtester")
 	testers.IgnoreNoCodecError = true
 	testers.IgnoreGaps = true
 	testers.IgnoreTimeDrift = true
@@ -93,8 +98,11 @@ func main() {
 		fmt.Println("Pause should be less than 5 min")
 		os.Exit(1)
 	}
-	var err error
+	if *analyzerServers == "" {
+		*analyzerServers = *apiServer
+	}
 	var fileName string
+	var err error
 
 	gctx, gcancel := context.WithCancel(context.Background()) // to be used as global parent context, in the future
 	defer gcancel()
@@ -155,6 +163,12 @@ func main() {
 	lapi.Init()
 	glog.Infof("Choosen server: %s", lapi.GetServer())
 
+	userAgent := model.AppName + "/" + model.Version
+	lanalyzers := testers.AnalyzerByRegion{}
+	for _, url := range strings.Split(*analyzerServers, ",") {
+		lanalyzers[url] = client.NewAnalyzer(url, *apiToken, userAgent, 0)
+	}
+
 	/*
 		sessionsx, err := lapi.GetSessions("1f770f0a-9177-49bd-a848-023abee7c09b")
 		if err != nil {
@@ -185,7 +199,7 @@ func main() {
 		start := time.Now()
 
 		for i := 0; i < *sim; i++ {
-			rt := recordtester.NewRecordTester(gctx, lapi, useForceURL, *useHttp, *testMP4)
+			rt := recordtester.NewRecordTester(gctx, lapi, lanalyzers, useForceURL, *useHttp, *testMP4, *testStreamHealth)
 			eses = append(eses, 0)
 			testers = append(testers, rt)
 			wg.Add(1)
@@ -196,7 +210,7 @@ func main() {
 				if les != 0 {
 					es = les
 				}
-				if err != nil {
+				if lerr != nil {
 					err = lerr
 				}
 				wg.Done()
@@ -219,7 +233,7 @@ func main() {
 	} else if *continuousTest > 0 {
 		metricServer := server.NewMetricsServer()
 		go metricServer.Start(gctx, *bind)
-		crt := recordtester.NewContinuousRecordTester(gctx, lapi, *pagerDutyIntegrationKey, *pagerDutyComponent, *useHttp, *testMP4)
+		crt := recordtester.NewContinuousRecordTester(gctx, lapi, lanalyzers, *pagerDutyIntegrationKey, *pagerDutyComponent, *pagerDutyLowUrgency, *useHttp, *testMP4, *testStreamHealth)
 		err := crt.Start(fileName, *testDuration, *pauseDuration, *continuousTest)
 		if err != nil {
 			glog.Warningf("Continuous test ended with err=%v", err)
@@ -228,7 +242,7 @@ func main() {
 		return
 	}
 	// just one stream
-	rt := recordtester.NewRecordTester(gctx, lapi, useForceURL, *useHttp, *testMP4)
+	rt := recordtester.NewRecordTester(gctx, lapi, lanalyzers, useForceURL, *useHttp, *testMP4, *testStreamHealth)
 	es, err := rt.Start(fileName, *testDuration, *pauseDuration)
 	exit(es, fileName, *fileArg, err)
 }

cmd/streamtester/streamtester.go

@@ -265,7 +265,7 @@ func main() {
 		}
 		msg := fmt.Sprintf(`Starting %s stream to %s, pulling from %s`, durs, *rtmpURL, *mediaURL)
 		messenger.SendMessage(msg)
-		sr2 := testers.NewStreamer2(gctx, *wowza, *mist, *save, true, true)
+		sr2 := testers.NewStreamer2(gctx, testers.Streamer2Options{*wowza, *mist, *save, true, true})
 		sr2.StartStreaming(fn, *rtmpURL, *mediaURL, *waitForTarget, *streamDuration)
 		if *wowza {
 			// let Wowza remove session
@@ -365,7 +365,7 @@ func main() {
 			msg := fmt.Sprintf(`Starting %s stream to %s`, dur, *rtmpURL)
 			glog.Info(msg)
 
-			sr2 := testers.NewStreamer2(gctx, *wowza, *mist, *save, true, true)
+			sr2 := testers.NewStreamer2(gctx, testers.Streamer2Options{*wowza, *mist, *save, true, true})
 			sr2.StartStreaming(fn, *rtmpURL, *mediaURL, *waitForTarget, *streamDuration)
 		}
 		err = lapi.DeleteStream(stream.ID)

go.mod

@@ -16,7 +16,7 @@ require (
 	github.com/gosuri/uiprogress v0.0.1
 	github.com/livepeer/joy4 v0.1.2-0.20210601043311-c1b885884cc7
 	github.com/livepeer/leaderboard-serverless v1.0.0
-	github.com/livepeer/livepeer-data v0.4.0
+	github.com/livepeer/livepeer-data v0.4.5-beta.3
 	github.com/livepeer/m3u8 v0.11.1
 	github.com/mattn/go-isatty v0.0.8 // indirect
 	github.com/patrickmn/go-cache v2.1.0+incompatible
@@ -25,6 +25,7 @@ require (
 	github.com/prometheus/client_golang v1.7.1
 	github.com/prometheus/common v0.15.0
 	github.com/rabbitmq/amqp091-go v1.1.0
+	github.com/rabbitmq/rabbitmq-stream-go-client v0.1.0-beta.0.20211027081212-fd5e6d497413 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.0
 	go.etcd.io/etcd/client/v3 v3.5.0-rc.0
 	go.opencensus.io v0.23.0

go.sum

@@ -117,6 +117,7 @@ github.com/coreos/pkg v0.0.0-20160727233714-3ac0863d7acf/go.mod h1:E3G3o1h8I7cfc
 github.com/cpuguy83/go-md2man/v2 v2.0.0-20190314233015-f79a8a8ca69d/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU=
 github.com/cpuguy83/go-md2man/v2 v2.0.0/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU=
 github.com/creack/pty v1.1.7/go.mod h1:lj5s0c3V2DBrqTV7llrYr5NG6My20zk30Fl46Y7DoTY=
+github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -143,6 +144,7 @@ github.com/form3tech-oss/jwt-go v3.2.2+incompatible h1:TcekIExNqud5crz4xD2pavyTg
 github.com/form3tech-oss/jwt-go v3.2.2+incompatible/go.mod h1:pbq4aXjuKjdthFRnoDwaVPLA+WlJuPGy+QneDUgJi2k=
 github.com/franela/goblin v0.0.0-20200105215937-c9ffbefa60db/go.mod h1:7dvUGVsVBjqR7JHJk0brhHOZYGmfBYOrK0ZhYMEtBr4=
 github.com/franela/goreq v0.0.0-20171204163338-bcd34c9993f8/go.mod h1:ZhphrRTfi2rbfLwlschooIH4+wKKDR4Pdxhh+TRoA20=
+github.com/frankban/quicktest v1.14.0/go.mod h1:NeW+ay9A/U67EYXNFA1nPE8e/tnQv/09mUdL/ijj8og=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
@@ -227,6 +229,7 @@ github.com/golang/protobuf v1.5.2 h1:ROPKBNFfQgOUMifHyP+KYbvpjbdoFNs+aK7DXlji0Tw
 github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
 github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
@@ -239,8 +242,9 @@ github.com/google/go-cmp v0.5.1/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.3/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.6 h1:BKbKCqvP6I+rmFHt06ZmyQtvB8xAkWdhFyr0ZUNZcxQ=
+github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-querystring v1.0.0 h1:Xkwi/a1rcvNg1PPYe5vI8GbeBY/jrVuDX5ASuANWTrk=
 github.com/google/go-querystring v1.0.0/go.mod h1:odCYkC5MyYFN7vkCjXpyrEuKhc/BUO6wN/zVPAxq5ck=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
@@ -338,25 +342,32 @@ github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvW
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/klauspost/compress v1.9.5/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
+github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/konsorten/go-windows-terminal-sequences v1.0.2/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/kr/fs v0.1.0/go.mod h1:FFnZGqtBN9Gxj7eW1uZ42v5BccTP0vu6NEaFoC2HwRg=
 github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
-github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+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 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 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=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/lib/pq v1.8.0/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
 github.com/lightstep/lightstep-tracer-common/golang/gogo v0.0.0-20190605223551-bc2310a04743/go.mod h1:qklhhLq1aX+mtWk9cPHPzaBjWImj5ULL6C7HFJtXQMM=
 github.com/lightstep/lightstep-tracer-go v0.18.1/go.mod h1:jlF1pusYV4pidLvZ+XD0UBX0ZE6WURAspgAczcDHrL4=
 github.com/livepeer/joy4 v0.1.2-0.20210601043311-c1b885884cc7 h1:gGU8CE90+YL2I+irPM6aQNfBBe8HYGyy7epnZYNo4QM=
 github.com/livepeer/joy4 v0.1.2-0.20210601043311-c1b885884cc7/go.mod h1:RDTLvmm/NJWjzuUpEDyIWmLTqSfpZEcnPnacG8sfh34=
 github.com/livepeer/leaderboard-serverless v1.0.0 h1:QqUxAEorjtfFxQO7tqTbxOLK+LFTJU27hvXf3/8YRZw=
 github.com/livepeer/leaderboard-serverless v1.0.0/go.mod h1:0H4gFyyTO3lXtqfRIwyRaWPuW0pBMXkNUDtIjHHniW0=
-github.com/livepeer/livepeer-data v0.4.0 h1:17rELy6TFoY+gjFgQtKbVNkn1jF7+QFD7aotMVlhnfI=
-github.com/livepeer/livepeer-data v0.4.0/go.mod h1:XRMZAh6E48fZS9M6i5JHZMfEUXuwaC3Zig5OSogROOM=
+github.com/livepeer/livepeer-data v0.4.5-beta h1:iyXhhpybqq6dceM0ldvFxHKivM3Rgrl3WktRuzSQlO4=
+github.com/livepeer/livepeer-data v0.4.5-beta/go.mod h1:KMmKC+o3b3C/n9NS5Zvp5SumxnzSW55pj45lJDb4b50=
+github.com/livepeer/livepeer-data v0.4.5-beta.2 h1:WN2CULcaixYSK2/S8wNJBFez3D1qpD3B6eR5ntZ5RIM=
+github.com/livepeer/livepeer-data v0.4.5-beta.2/go.mod h1:KMmKC+o3b3C/n9NS5Zvp5SumxnzSW55pj45lJDb4b50=
+github.com/livepeer/livepeer-data v0.4.5-beta.3 h1:ms9wKvze5zmFKRKevI1I+q3nCeV6kRFr6WKz9UgXA4Y=
+github.com/livepeer/livepeer-data v0.4.5-beta.3/go.mod h1:KMmKC+o3b3C/n9NS5Zvp5SumxnzSW55pj45lJDb4b50=
 github.com/livepeer/m3u8 v0.11.1 h1:VkUJzfNTyjy9mqsgp5JPvouwna8wGZMvd/gAfT5FinU=
 github.com/livepeer/m3u8 v0.11.1/go.mod h1:IUqAtwWPAG2CblfQa4SVzTQoDcEMPyfNOaBSxqHMS04=
 github.com/lyft/protoc-gen-validate v0.0.13/go.mod h1:XbGvPuh87YZc5TdIa2/I4pLk0QoUACkjt2znoq26NVQ=
@@ -400,6 +411,7 @@ github.com/nats-io/nats.go v1.9.1/go.mod h1:ZjDU1L/7fJ09jvUSRVBR2e7+RnLiiIQyqyzE
 github.com/nats-io/nkeys v0.1.0/go.mod h1:xpnFELMwJABBLVhffcfd1MZx6VsNRFpEugbxziKVo7w=
 github.com/nats-io/nkeys v0.1.3/go.mod h1:xpnFELMwJABBLVhffcfd1MZx6VsNRFpEugbxziKVo7w=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
+github.com/nbio/hitch v0.0.0-20200727225606-6dd6dd293f2b/go.mod h1:pCpSEQ+v0QJ68bwwrFH5BRKkPQyZUlKcxcYhIAlq48U=
 github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=
 github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
 github.com/nxadm/tail v1.4.8/go.mod h1:+ncqLTQzXmGhMZNUePPaPqPvBxHAIsmXswZKocGu+AU=
@@ -441,6 +453,7 @@ github.com/peterbourgon/ff/v2 v2.0.0 h1:lx0oYI5qr/FU1xnpNhQ+EZM04gKgn46jyYvGEEqB
 github.com/peterbourgon/ff/v2 v2.0.0/go.mod h1:xjwr+t+SjWm4L46fcj/D+Ap+6ME7+HqFzaP22pP5Ggk=
 github.com/pierrec/lz4 v1.0.2-0.20190131084431-473cd7ce01a1/go.mod h1:3/3N9NVKO0jef7pBehbT1qWhCMrIgbYNnFAZCqQ5LRc=
 github.com/pierrec/lz4 v2.0.5+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
+github.com/pierrec/lz4 v2.6.1+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
@@ -482,6 +495,7 @@ github.com/prometheus/procfs v0.1.3 h1:F0+tqvhOksq22sc6iCHF5WGlWjdwj92p0udFh1VFB
 github.com/prometheus/procfs v0.1.3/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4OA4YeYWdaU=
 github.com/rabbitmq/amqp091-go v1.1.0 h1:qx8cGMJha71/5t31Z+LdPLdPrkj/BvD38cqC3Bi1pNI=
 github.com/rabbitmq/amqp091-go v1.1.0/go.mod h1:ogQDLSOACsLPsIq0NpbtiifNZi2YOz0VTJ0kHRghqbM=
+github.com/rabbitmq/rabbitmq-stream-go-client v0.1.0-RC1/go.mod h1:31GXVWngt/bGfAPj/dQaVadf6YBmLjNhdoLHWoJEvh0=
 github.com/rabbitmq/rabbitmq-stream-go-client v0.1.0-beta.0.20211027081212-fd5e6d497413 h1:dHcuFg2/Ull4o4j2DZeu2cRFrlMEt+wERsXyajGG3vE=
 github.com/rabbitmq/rabbitmq-stream-go-client v0.1.0-beta.0.20211027081212-fd5e6d497413/go.mod h1:salB4WXuCH7GAMrciUT77dqEfkMvKFCpjqmx+vbonYk=
 github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
@@ -490,6 +504,8 @@ github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6L
 github.com/rogpeppe/go-internal v1.1.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.2.2/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
+github.com/rogpeppe/go-internal v1.6.1 h1:/FiVV8dS/e+YqF2JvO3yXRFbBLTIuSDkuC7aBOAvL+k=
+github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=
 github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts=
 github.com/samuel/go-zookeeper v0.0.0-20190923202752-2cc03de413da/go.mod h1:gi+0XIa01GRL2eRQVjQkKGqKF3SF9vZR/HnPullcV2E=