wip adds v2 metadatastore flags to dataapi

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

Hide file tree

Showing 6 changed files with 71 additions and 286 deletions.
diff --git a/disperser/cmd/dataapi/config.go b/disperser/cmd/dataapi/config.go
@@ -13,13 +13,14 @@ import (
 )
 
 type Config struct {
-	AwsClientConfig  aws.ClientConfig
-	BlobstoreConfig  blobstore.Config
-	EthClientConfig  geth.EthClientConfig
-	LoggerConfig     common.LoggerConfig
-	PrometheusConfig prometheus.Config
-	MetricsConfig    dataapi.MetricsConfig
-	ChainStateConfig thegraph.Config
+	AwsClientConfig   aws.ClientConfig
+	BlobstoreConfig   blobstore.Config
+	BlobstoreConfigV2 blobstore.Config
+	EthClientConfig   geth.EthClientConfig
+	LoggerConfig      common.LoggerConfig
+	PrometheusConfig  prometheus.Config
+	MetricsConfig     dataapi.MetricsConfig
+	ChainStateConfig  thegraph.Config
 
 	SocketAddr                   string
 	PrometheusApiAddr            string
@@ -34,6 +35,7 @@ type Config struct {
 	DisperserHostname  string
 	ChurnerHostname    string
 	BatcherHealthEndpt string
+	MetadataStoreV2    string
 }
 
 func NewConfig(ctx *cli.Context) (Config, error) {
@@ -47,6 +49,10 @@ func NewConfig(ctx *cli.Context) (Config, error) {
 			BucketName: ctx.GlobalString(flags.S3BucketNameFlag.Name),
 			TableName:  ctx.GlobalString(flags.DynamoTableNameFlag.Name),
 		},
+		BlobstoreConfigV2: blobstore.Config{
+			BucketName: ctx.GlobalString(flags.S3BucketNameFlag.Name),
+			TableName:  ctx.GlobalString(flags.DynamoMetadataStoreV2Flag.Name),
+		},
 		AwsClientConfig:               aws.ReadClientConfig(ctx, flags.FlagPrefix),
 		EthClientConfig:               ethClientConfig,
 		LoggerConfig:                  *loggerConfig,
@@ -72,6 +78,7 @@ func NewConfig(ctx *cli.Context) (Config, error) {
 		ChurnerHostname:    ctx.GlobalString(flags.ChurnerHostnameFlag.Name),
 		BatcherHealthEndpt: ctx.GlobalString(flags.BatcherHealthEndptFlag.Name),
 		ChainStateConfig:   thegraph.ReadCLIConfig(ctx),
+		MetadataStoreV2:    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"
@@ -91,13 +92,19 @@ 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(
+		sharedStorage     = blobstore.NewSharedStorage(
+			config.BlobstoreConfig.BucketName,
+			s3Client,
+			blobMetadataStore,
+			logger,
+		)
+		blobMetadataStoreV2 = v2blobstore.NewBlobMetadataStore(dynamoClient, logger, config.BlobstoreConfigV2.TableName)
+		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 +114,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,7 +24,9 @@ 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"
 	"github.com/gin-gonic/gin"
@@ -196,6 +198,7 @@ type (
 		allowOrigins      []string
 		logger            logging.Logger
 		blobstore         disperser.BlobStore
+		blobstoreV2       *v2blobstore.BlobMetadataStore
 		promClient        PrometheusClient
 		subgraphClient    SubgraphClient
 		transactor        core.Reader
@@ -214,6 +217,7 @@ type (
 func NewServer(
 	config Config,
 	blobstore disperser.BlobStore,
+	blobstoreV2 *v2blobstore.BlobMetadataStore,
 	promClient PrometheusClient,
 	subgraphClient SubgraphClient,
 	transactor core.Reader,
@@ -245,6 +249,7 @@ func NewServer(
 		socketAddr:                config.SocketAddr,
 		allowOrigins:              config.AllowOrigins,
 		blobstore:                 blobstore,
+		blobstoreV2:               blobstoreV2,
 		promClient:                promClient,
 		subgraphClient:            subgraphClient,
 		transactor:                transactor,