wip adds v2 metadatastore flags to dataapi

Layr-Labs · Nov 20, 2024 · b4b5cc8 · b4b5cc8
1 parent afd5894
commit b4b5cc8
Show file tree

Hide file tree

Showing 6 changed files with 62 additions and 288 deletions.
diff --git a/disperser/cmd/dataapi/config.go b/disperser/cmd/dataapi/config.go
@@ -31,9 +31,10 @@ type Config struct {
 	BLSOperatorStateRetrieverAddr string
 	EigenDAServiceManagerAddr     string
 
-	DisperserHostname  string
-	ChurnerHostname    string
-	BatcherHealthEndpt string
+	DisperserHostname   string
+	ChurnerHostname     string
+	BatcherHealthEndpt  string
+	BlobMetadataStoreV2 string
 }
 
 func NewConfig(ctx *cli.Context) (Config, error) {
@@ -68,10 +69,11 @@ func NewConfig(ctx *cli.Context) (Config, error) {
 			HTTPPort:      ctx.GlobalString(flags.MetricsHTTPPort.Name),
 			EnableMetrics: ctx.GlobalBool(flags.EnableMetricsFlag.Name),
 		},
-		DisperserHostname:  ctx.GlobalString(flags.DisperserHostnameFlag.Name),
-		ChurnerHostname:    ctx.GlobalString(flags.ChurnerHostnameFlag.Name),
-		BatcherHealthEndpt: ctx.GlobalString(flags.BatcherHealthEndptFlag.Name),
-		ChainStateConfig:   thegraph.ReadCLIConfig(ctx),
+		DisperserHostname:   ctx.GlobalString(flags.DisperserHostnameFlag.Name),
+		ChurnerHostname:     ctx.GlobalString(flags.ChurnerHostnameFlag.Name),
+		BatcherHealthEndpt:  ctx.GlobalString(flags.BatcherHealthEndptFlag.Name),
+		ChainStateConfig:    thegraph.ReadCLIConfig(ctx),
+		BlobMetadataStoreV2: ctx.GlobalString(flags.DynamoMetadataStoreV2Flag.Name),
 	}
 	return config, nil
 }
diff --git a/disperser/cmd/dataapi/flags/flags.go b/disperser/cmd/dataapi/flags/flags.go
@@ -20,6 +20,12 @@ var (
 		Required: true,
 		EnvVar:   common.PrefixEnvVar(envVarPrefix, "DYNAMO_TABLE_NAME"),
 	}
+	DynamoMetadataStoreV2Flag = cli.StringFlag{
+		Name:     common.PrefixFlag(FlagPrefix, "dynamo-metadata-store-v2"),
+		Usage:    "Name of the dynamo table to store v2 blob metadata",
+		Required: false,
+		EnvVar:   common.PrefixEnvVar(envVarPrefix, "DYNAMO_METADATA_STORE_V2"),
+	}
 	S3BucketNameFlag = cli.StringFlag{
 		Name:     common.PrefixFlag(FlagPrefix, "s3-bucket-name"),
 		Usage:    "Name of the bucket to store blobs",
@@ -136,6 +142,7 @@ var (
 
 var requiredFlags = []cli.Flag{
 	DynamoTableNameFlag,
+	DynamoMetadataStoreV2Flag,
 	SocketAddrFlag,
 	S3BucketNameFlag,
 	SubgraphApiBatchMetadataAddrFlag,

diff --git a/disperser/cmd/dataapi/main.go b/disperser/cmd/dataapi/main.go
@@ -16,6 +16,7 @@ import (
 	"github.com/Layr-Labs/eigenda/core/thegraph"
 	"github.com/Layr-Labs/eigenda/disperser/cmd/dataapi/flags"
 	"github.com/Layr-Labs/eigenda/disperser/common/blobstore"
+	v2blobstore "github.com/Layr-Labs/eigenda/disperser/common/v2/blobstore"
 	"github.com/Layr-Labs/eigenda/disperser/dataapi"
 	"github.com/Layr-Labs/eigenda/disperser/dataapi/prometheus"
 	"github.com/Layr-Labs/eigenda/disperser/dataapi/subgraph"
@@ -89,15 +90,16 @@ func RunDataApi(ctx *cli.Context) error {
 	}
 
 	var (
-		promClient        = dataapi.NewPrometheusClient(promApi, config.PrometheusConfig.Cluster)
-		blobMetadataStore = blobstore.NewBlobMetadataStore(dynamoClient, logger, config.BlobstoreConfig.TableName, 0)
-		sharedStorage     = blobstore.NewSharedStorage(config.BlobstoreConfig.BucketName, s3Client, blobMetadataStore, logger)
-		subgraphApi       = subgraph.NewApi(config.SubgraphApiBatchMetadataAddr, config.SubgraphApiOperatorStateAddr)
-		subgraphClient    = dataapi.NewSubgraphClient(subgraphApi, logger)
-		chainState        = coreeth.NewChainState(tx, client)
-		indexedChainState = thegraph.MakeIndexedChainState(config.ChainStateConfig, chainState, logger)
-		metrics           = dataapi.NewMetrics(blobMetadataStore, config.MetricsConfig.HTTPPort, logger)
-		server            = dataapi.NewServer(
+		promClient          = dataapi.NewPrometheusClient(promApi, config.PrometheusConfig.Cluster)
+		blobMetadataStore   = blobstore.NewBlobMetadataStore(dynamoClient, logger, config.BlobstoreConfig.TableName, 0)
+		sharedStorage       = blobstore.NewSharedStorage(config.BlobstoreConfig.BucketName, s3Client, blobMetadataStore, logger)
+		blobMetadataStoreV2 = v2blobstore.NewBlobMetadataStore(dynamoClient, logger, config.BlobMetadataStoreV2)
+		subgraphApi         = subgraph.NewApi(config.SubgraphApiBatchMetadataAddr, config.SubgraphApiOperatorStateAddr)
+		subgraphClient      = dataapi.NewSubgraphClient(subgraphApi, logger)
+		chainState          = coreeth.NewChainState(tx, client)
+		indexedChainState   = thegraph.MakeIndexedChainState(config.ChainStateConfig, chainState, logger)
+		metrics             = dataapi.NewMetrics(blobMetadataStore, config.MetricsConfig.HTTPPort, logger)
+		server              = dataapi.NewServer(
 			dataapi.Config{
 				ServerMode:         config.ServerMode,
 				SocketAddr:         config.SocketAddr,
@@ -107,6 +109,7 @@ func RunDataApi(ctx *cli.Context) error {
 				BatcherHealthEndpt: config.BatcherHealthEndpt,
 			},
 			sharedStorage,
+			blobMetadataStoreV2,
 			promClient,
 			subgraphClient,
 			tx,

diff --git a/disperser/common/blobstore/blob_metadata_store.go b/disperser/common/blobstore/blob_metadata_store.go
@@ -46,6 +46,16 @@ func NewBlobMetadataStore(dynamoDBClient commondynamodb.Client, logger logging.L
 	}
 }
 
+func NewBlobMetadataStoreV2(dynamoDBClient commondynamodb.Client, logger logging.Logger, tableName string, ttl time.Duration) *BlobMetadataStore {
+	logger.Debugf("creating blob metadata store v2 with table %s with TTL: %s", tableName, ttl)
+	return &BlobMetadataStore{
+		dynamoDBClient: dynamoDBClient,
+		logger:         logger.With("component", "BlobMetadataStoreV2"),
+		tableName:      tableName,
+		ttl:            ttl,
+	}
+}
+
 func (s *BlobMetadataStore) QueueNewBlobMetadata(ctx context.Context, blobMetadata *disperser.BlobMetadata) error {
 	item, err := MarshalBlobMetadata(blobMetadata)
 	if err != nil {

diff --git a/disperser/dataapi/server.go b/disperser/dataapi/server.go
@@ -24,6 +24,7 @@ import (
 
 	"github.com/Layr-Labs/eigenda/disperser"
 	"github.com/Layr-Labs/eigenda/disperser/common/semver"
+	v2blobstore "github.com/Layr-Labs/eigenda/disperser/common/v2/blobstore"
 	"github.com/Layr-Labs/eigenda/disperser/dataapi/docs"
 	"github.com/gin-contrib/cors"
 	"github.com/gin-contrib/logger"
@@ -196,6 +197,7 @@ type (
 		allowOrigins      []string
 		logger            logging.Logger
 		blobstore         disperser.BlobStore
+		blobstoreV2       *v2blobstore.BlobMetadataStore
 		promClient        PrometheusClient
 		subgraphClient    SubgraphClient
 		transactor        core.Reader
@@ -214,6 +216,7 @@ type (
 func NewServer(
 	config Config,
 	blobstore disperser.BlobStore,
+	blobstoreV2 *v2blobstore.BlobMetadataStore,
 	promClient PrometheusClient,
 	subgraphClient SubgraphClient,
 	transactor core.Reader,
@@ -245,6 +248,7 @@ func NewServer(
 		socketAddr:                config.SocketAddr,
 		allowOrigins:              config.AllowOrigins,
 		blobstore:                 blobstore,
+		blobstoreV2:               blobstoreV2,
 		promClient:                promClient,
 		subgraphClient:            subgraphClient,
 		transactor:                transactor,