771 read traffic generator

inabox/deploy/utils.go

@@ -163,8 +163,8 @@ func execForgeScript(script, privateKey string, deployer *ContractDeployer, extr
 	log.Println("Executing forge script with command: ", cmd.String())
 	err := cmd.Run()
 	if err != nil {
-		log.Print(fmt.Sprint(err) + ": " + stderr.String())
-		log.Panicf("Failed to execute forge script. Err: %s", err)
+		log.Panicf("Failed to execute forge script: %s\n Err: %s\n--- std out ---\n%s\n--- std err ---\n%s\n",
+			cmd, err, out.String(), stderr.String())
 	} else {
 		log.Print(out.String())
 	}

tools/traffic/Makefile

@@ -15,3 +15,37 @@ run: build
 	TRAFFIC_GENERATOR_DATA_SIZE=1000 \
 	TRAFFIC_GENERATOR_RANDOMIZE_BLOBS=true \
 	./bin/server
+
+build2: clean
+	# cd ../.. && make protoc
+	go mod tidy
+	go build -o ./bin/server2 ./cmd2
+
+run2: build2
+	TRAFFIC_GENERATOR_LOG_FORMAT=text \
+	TRAFFIC_GENERATOR_DISPERSER_HOSTNAME=localhost \
+	TRAFFIC_GENERATOR_DISPERSER_PORT=32003 \
+	RETRIEVER_HOSTNAME=localhost \
+	RETRIEVER_GRPC_PORT=32003 \
+	RETRIEVER_BLS_OPERATOR_STATE_RETRIVER=0x5f3f1dBD7B74C6B46e8c44f98792A1dAf8d69154 \
+	RETRIEVER_EIGENDA_SERVICE_MANAGER=0x851356ae760d987E095750cCeb3bC6014560891C \
+	RETRIEVER_TIMEOUT=10s \
+	TRAFFIC_GENERATOR_TIMEOUT=10s \
+	TRAFFIC_GENERATOR_NUM_WRITE_INSTANCES=1 \
+	TRAFFIC_GENERATOR_WRITE_REQUEST_INTERVAL=1s \
+	TRAFFIC_GENERATOR_DATA_SIZE=1000 \
+	TRAFFIC_GENERATOR_RANDOMIZE_BLOBS=true \
+	TRAFFIC_GENERATOR_SIGNER_PRIVATE_KEY_HEX=AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA \
+	TRAFFIC_GENERATOR_PRIVATE_KEY=AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA \
+	TRAFFIC_GENERATOR_METRICS_HTTP_PORT=9101 \
+	TRAFFIC_GENERATOR_GRAPH_URL=http://localhost:8000/subgraphs/name/Layr-Labs/eigenda-operator-state \
+	TRAFFIC_GENERATOR_THE_GRAPH_URL=http://localhost:8000/subgraphs/name/Layr-Labs/eigenda-operator-state \
+	TRAFFIC_GENERATOR_G1_PATH=../../inabox/resources/kzg/g1.point \
+	TRAFFIC_GENERATOR_G2_PATH=../../inabox/resources/kzg/g2.point \
+	TRAFFIC_GENERATOR_CACHE_PATH=../../inabox/resources/kzg/SRSTables \
+	TRAFFIC_GENERATOR_SRS_ORDER=3000 \
+	TRAFFIC_GENERATOR_SRS_LOAD=3000 \
+	TRAFFIC_GENERATOR_NUM_WORKERS=4 \
+	TRAFFIC_GENERATOR_CHAIN_RPC=http://localhost:8545 \
+	TRAFFIC_GENERATOR_NUM_RETRIES=2 \
+	./bin/server2

tools/traffic/cmd2/main.go

@@ -0,0 +1,44 @@
+package main
+
+import (
+	"fmt"
+	"log"
+	"os"
+
+	"github.com/Layr-Labs/eigenda/tools/traffic"
+	"github.com/Layr-Labs/eigenda/tools/traffic/config"
+	"github.com/urfave/cli"
+)
+
+var (
+	version   = ""
+	gitCommit = ""
+	gitDate   = ""
+)
+
+func main() {
+	app := cli.NewApp()
+	app.Version = fmt.Sprintf("%s-%s-%s", version, gitCommit, gitDate)
+	app.Name = "da-traffic-generator"
+	app.Usage = "EigenDA Traffic Generator"
+	app.Description = "Service for generating traffic to EigenDA disperser"
+	app.Flags = config.Flags
+	app.Action = trafficGeneratorMain
+	if err := app.Run(os.Args); err != nil {
+		log.Fatalf("application failed: %v", err)
+	}
+}
+
+func trafficGeneratorMain(ctx *cli.Context) error {
+	generatorConfig, err := config.NewConfig(ctx)
+	if err != nil {
+		return err
+	}
+
+	generator, err := traffic.NewTrafficGeneratorV2(generatorConfig)
+	if err != nil {
+		panic(fmt.Sprintf("failed to create new traffic generator\n%s", err))
+	}
+
+	return generator.Start()
+}

tools/traffic/generator_v2.go

@@ -0,0 +1,217 @@
+package traffic
+
+import (
+	"context"
+	"fmt"
+	"github.com/Layr-Labs/eigenda/common/geth"
+	"github.com/Layr-Labs/eigenda/core/auth"
+	"github.com/Layr-Labs/eigenda/core/eth"
+	"github.com/Layr-Labs/eigenda/core/thegraph"
+	"github.com/Layr-Labs/eigenda/encoding/kzg/verifier"
+	retrivereth "github.com/Layr-Labs/eigenda/retriever/eth"
+	"github.com/Layr-Labs/eigenda/tools/traffic/config"
+	"github.com/Layr-Labs/eigenda/tools/traffic/metrics"
+	"github.com/Layr-Labs/eigenda/tools/traffic/table"
+	"github.com/Layr-Labs/eigenda/tools/traffic/workers"
+	"github.com/Layr-Labs/eigensdk-go/logging"
+	gethcommon "github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/log"
+	"os"
+	"os/signal"
+	"sync"
+	"syscall"
+	"time"
+
+	"github.com/Layr-Labs/eigenda/api/clients"
+	"github.com/Layr-Labs/eigenda/common"
+	"github.com/Layr-Labs/eigenda/core"
+)
+
+// Generator simulates read/write traffic to the DA service.
+//
+//	┌------------┐                                           ┌------------┐
+//	|   writer   |-┐             ┌----------------┐          |   reader   |-┐
+//	└------------┘ |-┐  -------> | status tracker | -------> └------------┘ |-┐
+//	  └------------┘ |           └----------------┘            └------------┘ |
+//	    └------------┘                                           └------------┘
+//
+// The traffic generator is built from three principal components: one or more writers
+// that write blobs, a statusTracker that polls the dispenser service until blobs are confirmed,
+// and one or more readers that read blobs.
+//
+// When a writer finishes writing a blob, it sends information about that blob to the statusTracker.
+// When the statusTracker observes that a blob has been confirmed, it sends information about the blob
+// to the readers. The readers only attempt to read blobs that have been confirmed by the statusTracker.
+type Generator struct {
+	ctx              *context.Context
+	cancel           *context.CancelFunc
+	waitGroup        *sync.WaitGroup
+	generatorMetrics metrics.Metrics
+	logger           *logging.Logger
+	disperserClient  clients.DisperserClient
+	eigenDAClient    *clients.EigenDAClient
+	config           *config.Config
+
+	writers       []*workers.BlobWriter
+	statusTracker *workers.BlobStatusTracker
+	readers       []*workers.BlobReader
+}
+
+func NewTrafficGeneratorV2(config *config.Config) (*Generator, error) {
+	logger, err := common.NewLogger(config.LoggingConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	var signer core.BlobRequestSigner
+	if config.EigenDAClientConfig.SignerPrivateKeyHex != "" {
+		signer = auth.NewLocalBlobRequestSigner(config.EigenDAClientConfig.SignerPrivateKeyHex)
+	}
+
+	logger2 := log.NewLogger(log.NewTerminalHandler(os.Stderr, true))
+	client, err := clients.NewEigenDAClient(logger2, *config.EigenDAClientConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	waitGroup := sync.WaitGroup{}
+
+	generatorMetrics := metrics.NewMetrics(config.MetricsHTTPPort, logger)
+
+	blobTable := table.NewBlobStore()
+
+	unconfirmedKeyChannel := make(chan *workers.UnconfirmedKey, 100)
+
+	disperserClient := clients.NewDisperserClient(config.DisperserClientConfig, signer)
+	statusVerifier := workers.NewBlobStatusTracker(
+		&ctx,
+		&waitGroup,
+		logger,
+		&config.WorkerConfig,
+		unconfirmedKeyChannel,
+		blobTable,
+		disperserClient,
+		generatorMetrics)
+
+	writers := make([]*workers.BlobWriter, 0)
+	for i := 0; i < int(config.WorkerConfig.NumWriteInstances); i++ {
+		writer := workers.NewBlobWriter(
+			&ctx,
+			&waitGroup,
+			logger,
+			&config.WorkerConfig,
+			disperserClient,
+			unconfirmedKeyChannel,
+			generatorMetrics)
+		writers = append(writers, &writer)
+	}
+
+	retriever, chainClient := buildRetriever(config)
+
+	readers := make([]*workers.BlobReader, 0)
+	for i := 0; i < int(config.WorkerConfig.NumReadInstances); i++ {
+		reader := workers.NewBlobReader(
+			&ctx,
+			&waitGroup,
+			logger,
+			&config.WorkerConfig,
+			retriever,
+			chainClient,
+			blobTable,
+			generatorMetrics)
+		readers = append(readers, &reader)
+	}
+
+	return &Generator{
+		ctx:              &ctx,
+		cancel:           &cancel,
+		waitGroup:        &waitGroup,
+		generatorMetrics: generatorMetrics,
+		logger:           &logger,
+		disperserClient:  clients.NewDisperserClient(config.DisperserClientConfig, signer),
+		eigenDAClient:    client,
+		config:           config,
+		writers:          writers,
+		statusTracker:    &statusVerifier,
+		readers:          readers,
+	}, nil
+}
+
+// buildRetriever creates a retriever client for the traffic generator.
+func buildRetriever(config *config.Config) (clients.RetrievalClient, retrivereth.ChainClient) {
+	loggerConfig := common.DefaultLoggerConfig()
+
+	logger, err := common.NewLogger(loggerConfig)
+	if err != nil {
+		panic(fmt.Sprintf("Unable to instantiate logger: %s", err))
+	}
+
+	gethClient, err := geth.NewMultiHomingClient(config.RetrievalClientConfig.EthClientConfig, gethcommon.Address{}, logger)
+	if err != nil {
+		panic(fmt.Sprintf("Unable to instantiate geth client: %s", err))
+	}
+
+	tx, err := eth.NewTransactor(
+		logger,
+		gethClient,
+		config.RetrievalClientConfig.BLSOperatorStateRetrieverAddr,
+		config.RetrievalClientConfig.EigenDAServiceManagerAddr)
+	if err != nil {
+		panic(fmt.Sprintf("Unable to instantiate transactor: %s", err))
+	}
+
+	cs := eth.NewChainState(tx, gethClient)
+
+	chainState := thegraph.MakeIndexedChainState(*config.TheGraphConfig, cs, logger)
+
+	var assignmentCoordinator core.AssignmentCoordinator = &core.StdAssignmentCoordinator{}
+
+	nodeClient := clients.NewNodeClient(config.NodeClientTimeout)
+
+	v, err := verifier.NewVerifier(&config.RetrievalClientConfig.EncoderConfig, true)
+	if err != nil {
+		panic(fmt.Sprintf("Unable to build statusTracker: %s", err))
+	}
+
+	retriever, err := clients.NewRetrievalClient(
+		logger,
+		chainState,
+		assignmentCoordinator,
+		nodeClient,
+		v,
+		config.RetrievalClientConfig.NumConnections)
+
+	if err != nil {
+		panic(fmt.Sprintf("Unable to build retriever: %s", err))
+	}
+
+	chainClient := retrivereth.NewChainClient(gethClient, logger)
+
+	return retriever, chainClient
+}
+
+// Start instantiates goroutines that generate read/write traffic, continues until a SIGTERM is observed.
+func (generator *Generator) Start() error {
+
+	generator.generatorMetrics.Start()
+	generator.statusTracker.Start()
+
+	for _, writer := range generator.writers {
+		writer.Start()
+		time.Sleep(generator.config.InstanceLaunchInterval)
+	}
+
+	for _, reader := range generator.readers {
+		reader.Start()
+		time.Sleep(generator.config.InstanceLaunchInterval)
+	}
+
+	signals := make(chan os.Signal, 1)
+	signal.Notify(signals, os.Interrupt, syscall.SIGTERM)
+	<-signals
+
+	(*generator.cancel)()
+	generator.waitGroup.Wait()
+	return nil
+}