recordtester: Use random serf memeber for playback if enabled.

.editorconfig

@@ -14,6 +14,10 @@ end_of_line = lf
 [Makefile]
 indent_style = tab
 
+[Dockerfile*]
+indent_style = tab
+indent_size = 8
+
 # YAML Files
 [*.{yml,yaml,sh}]
 indent_size = 2

cmd/recordtester/recordtester.go

@@ -16,6 +16,7 @@ import (
 	"time"
 
 	"github.com/golang/glog"
+	serfClient "github.com/hashicorp/serf/client"
 	api "github.com/livepeer/go-api-client"
 	"github.com/livepeer/joy4/format"
 	"github.com/livepeer/livepeer-data/pkg/client"
@@ -36,6 +37,23 @@ func init() {
 	rand.Seed(time.Now().UnixNano())
 }
 
+func getSerfMembers(serfRPCAddr string) ([]serfClient.Member, error) {
+	if serfRPCAddr == "" {
+		return nil, nil
+	}
+	rpcClient, err := serfClient.NewRPCClient(serfRPCAddr)
+	if err != nil {
+		glog.Error(err)
+		return nil, err
+	}
+	members, err := rpcClient.Members()
+	if err != nil {
+		glog.Error(err)
+		return nil, err
+	}
+	return members, nil
+}
+
 func main() {
 	flag.Set("logtostderr", "true")
 	vFlag := flag.Lookup("v")
@@ -71,6 +89,11 @@ func main() {
 	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")
 
+	serfRPCAddr := fs.String("serf-rpc-addr", "", "Serf RPC address for fetching serf members")
+	useSerf := fs.Bool("use-serf", false, "Use serf playback URLs")
+	useRandomSerfMember := fs.Bool("random-serf-member", false, "Use a random member from serf member list")
+	serfPullCount := fs.Int("pull-count", 1, "Number of serf nodes to pull playback from (requires `--use-serf`)")
+
 	_ = fs.String("config", "", "config file (optional)")
 
 	ff.Parse(fs, os.Args[1:],
@@ -125,10 +148,13 @@ func main() {
 	// model.ProfilesNum = int(*profiles)
 
 	if *testDuration == 0 {
-		glog.Fatalf("-test-dur should be specified")
+		glog.Fatal("--test-dur should be specified")
 	}
 	if *apiToken == "" {
-		glog.Fatalf("-api-token should be specified")
+		glog.Fatal("--api-token should be specified")
+	}
+	if *useSerf && *serfRPCAddr == "" {
+		glog.Fatal("--serf-rpc-addr needed with --use-serf option")
 	}
 
 	if fileName, err = utils.GetFile(*fileArg, strings.ReplaceAll(hostName, ".", "_")); err != nil {
@@ -143,10 +169,21 @@ func main() {
 	var ingest *api.Ingest
 	if *ingestStr != "" {
 		if err := json.Unmarshal([]byte(*ingestStr), &ingest); err != nil {
-			glog.Fatalf("Error parsing -ingest argument: %v", err)
+			glog.Fatalf("Error parsing --ingest argument: %v", err)
 		}
 	}
 
+	serfMembers, err := getSerfMembers(*serfRPCAddr)
+	if err != nil {
+		glog.Fatalf("failed to process serf members: %v", err)
+	}
+	serfOptions := recordtester.SerfOptions{
+		UseSerf:          *useSerf,
+		SerfMembers:      serfMembers,
+		RandomSerfMember: *useRandomSerfMember,
+		SerfPullCount:    *serfPullCount,
+	}
+
 	var lapi *api.Client
 	var createdAPIStreams []string
 	cleanup := func(fn, fa string) {
@@ -232,7 +269,7 @@ func main() {
 		start := time.Now()
 
 		for i := 0; i < *sim; i++ {
-			rt := recordtester.NewRecordTester(gctx, rtOpts)
+			rt := recordtester.NewRecordTester(gctx, rtOpts, serfOptions)
 			eses = append(eses, 0)
 			testers = append(testers, rt)
 			wg.Add(1)
@@ -275,7 +312,7 @@ func main() {
 					PagerDutyLowUrgency:     *pagerDutyLowUrgency,
 					RecordTesterOptions:     rtOpts,
 				}
-				crt := recordtester.NewContinuousRecordTester(egCtx, crtOpts)
+				crt := recordtester.NewContinuousRecordTester(egCtx, crtOpts, serfOptions)
 				return crt.Start(fileName, *testDuration, *pauseDuration, *continuousTest)
 			})
 		}
@@ -298,7 +335,7 @@ func main() {
 		return
 	}
 	// just one stream
-	rt := recordtester.NewRecordTester(gctx, rtOpts)
+	rt := recordtester.NewRecordTester(gctx, rtOpts, serfOptions)
 	es, err := rt.Start(fileName, *testDuration, *pauseDuration)
 	exit(es, fileName, *fileArg, err)
 }

docker/Dockerfile.record-tester

@@ -21,7 +21,7 @@ FROM --platform=$TARGETPLATFORM	debian:stretch-slim
 
 WORKDIR	/root
 
-RUN     apt update && \
+RUN	apt update && \
 	apt install -yqq ca-certificates curl && \
 	apt clean
 

go.mod

@@ -97,7 +97,7 @@ require (
 	golang.org/x/mod v0.5.1 // indirect
 	golang.org/x/oauth2 v0.0.0-20210427180440-81ed05c6b58c // indirect
 	golang.org/x/sync v0.0.0-20220601150217-0de741cfad7f
-	golang.org/x/sys v0.0.0-20220319134239-a9b59b0215f8 // indirect
+	golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10 // indirect
 	golang.org/x/tools v0.1.9 // indirect
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
@@ -111,9 +111,25 @@ require (
 
 require (
 	github.com/eventials/go-tus v0.0.0-20220610120217-05d0564bb571 // indirect
+	github.com/hashicorp/serf v0.10.1
 	github.com/tomnomnom/linkheader v0.0.0-20180905144013-02ca5825eb80 // indirect
 )
 
+require (
+	github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da // indirect
+	github.com/google/btree v1.0.0 // indirect
+	github.com/hashicorp/errwrap v1.0.0 // indirect
+	github.com/hashicorp/go-immutable-radix v1.0.0 // indirect
+	github.com/hashicorp/go-msgpack v0.5.3 // indirect
+	github.com/hashicorp/go-multierror v1.1.0 // indirect
+	github.com/hashicorp/go-sockaddr v1.0.0 // indirect
+	github.com/hashicorp/golang-lru v0.5.5-0.20210104140557-80c98217689d // indirect
+	github.com/hashicorp/logutils v1.0.0 // indirect
+	github.com/hashicorp/memberlist v0.5.0 // indirect
+	github.com/miekg/dns v1.1.41 // indirect
+	github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 // indirect
+)
+
 exclude github.com/gosuri/uilive v0.0.4 // cause memory corruption
 
 // replace github.com/livepeer/joy4 => /Users/dark/projects/livepeer/joy4